Posted inDatabase

Chạy InfluxDB trên Docker: Giám sát Time-Series bền bỉ cho luồng dữ liệu IoT tốc độ cao

Database tutorial - IT technology blog
Database tutorial - IT technology blog

Vấn đề thực tế: Khi cơ sở dữ liệu quan hệ nghẽn vì luồng dữ liệu cảm biến

Năm ngoái, tôi hỗ trợ một khách hàng sản xuất xây dựng pipeline nạp dữ liệu (ingestion pipeline) cho 500 đầu dò nhiệt độ và độ rung IoT công nghiệp. Mỗi thiết bị gửi dữ liệu nhiệt độ, độ rung 3 trục và công suất tiêu thụ định kỳ 2 giây một lần. Con số này lên tới 250 payload mỗi giây—khoảng 21,6 triệu bản ghi ghi xuống đĩa mỗi 24 giờ.

Ban đầu, chúng tôi đẩy thẳng luồng dữ liệu vào một instance PostgreSQL 14 đã được tinh chỉnh, chạy trên máy ảo 8 core, 32 GB RAM. Trong mười ngày đầu, hiệu năng vẫn duy trì ổn định. Nhưng sang tuần thứ ba, các truy vấn tính giá trị trung bình trượt 30 ngày bắt đầu chạm trần và liên tục bị timeout sau 60 giây.

Disk I/O chạm ngưỡng bão hòa 98%. Các chỉ mục B-Tree phức hợp của bảng phình to vượt mức 45 GB—lớn hơn nhiều so với dung lượng RAM khả dụng. Tỷ lệ buffer cache miss tăng vọt, các lượt ghi mới bị dồn ứ lại ở Redis, và dàn ingestion worker cuối cùng cạn kiệt bộ nhớ (OOM) rồi dừng hoạt động.

Postgres là một database engine xuất sắc. MySQL và MongoDB cũng vậy. Thế nhưng, việc đổ dữ liệu đo từ xa (telemetry) cảm biến liên tục theo kiểu chỉ ghi nối tiếp (append-only) vào một kho dữ liệu quan hệ đa dụng sẽ kích hoạt những nút thắt cổ chai cấu trúc ở tầng đĩa mà việc tối ưu chỉ mục đơn thuần hoàn toàn không thể giải quyết.

Phân tích nguyên nhân gốc rễ: Tại sao các engine truyền thống lại nghẽn trước luồng ghi liên tục

Vấn đề cốt lõi bắt nguồn từ cách các engine đa dụng sắp xếp dữ liệu trên đĩa:

  • Nghẽn do duy trì chỉ mục (Index Maintenance Thrashing): Mỗi bản ghi insert mới đều phải cập nhật lại chỉ mục bảng. Khi kích thước bảng và chỉ mục vượt quá RAM, tầng lưu trữ phải dành phần lớn thời gian để đọc và cập nhật các trang B-Tree nằm rải rác khắp đĩa. Throughput giảm sút nghiêm trọng.
  • Khuếch đại theo định dạng dòng (Row-Oriented Amplification): Các hệ thống quan hệ ghi toàn bộ tuple dòng liên tiếp nhau. Để tính nhiệt độ trung bình trong 6 giờ của các cảm biến nhà xưởng, engine buộc phải quét hàng megabyte dữ liệu không liên quan như độ rung, công suất và trạng thái từ đĩa. Cách tổ chức này gây lãng phí lớn bộ nhớ và băng thông bus.
  • Dọn dẹp dữ liệu cũ cực kỳ nặng nề: Việc xóa dữ liệu cảm biến vượt quá chu kỳ lưu trữ (retention window) 90 ngày thường phải dùng các lệnh DELETE theo batch. Với Postgres hoặc MySQL, xóa 50 triệu dòng sẽ để lại dead tuples. Điều này dẫn tới phân mảnh bảng, làm phình to Write-Ahead Log (WAL), và buộc các tiến trình autovacuum phải chạy liên tục, gây nghẽn toàn bộ luồng ghi dữ liệu.

So sánh kiến trúc: Đa dụng vs Chuyên biệt cho Time-Series

Các đội ngũ xử lý luồng số liệu thường cân nhắc ba giải pháp:

1. Cơ sở dữ liệu quan hệ (PostgreSQL / TimescaleDB)

Postgres tiêu chuẩn mang lại cam kết ACID vững chắc và khả năng join dữ liệu quan hệ, nhưng các bảng thuần cần phải phân vùng thủ công triệt để (declarative range partitioning theo thời gian). TimescaleDB khắc phục nhiều hạn chế này nhờ hypertable. Dù vậy, bạn vẫn phải trả một khoản chi phí bộ nhớ đáng kể cho write buffer và cấu trúc chỉ mục.

2. Document Store (MongoDB)

MongoDB mở rộng quy mô ghi theo chiều ngang rất tốt và xử lý linh hoạt các payload không đồng nhất mà không cần migration bảng trước. Mặc dù vậy, việc downsample theo thời gian đòi hỏi các aggregation pipeline tùy chỉnh hoặc background worker riêng biệt. Những tiến trình này sẽ cạnh tranh trực tiếp CPU và bộ nhớ đệm WiredTiger với các luồng nạp dữ liệu.

3. Time-Series Engine chuyên biệt (InfluxDB)

InfluxDB sử dụng Time-Structured Merge Tree (TSM) engine kết hợp với định dạng lưu trữ theo cột trên đĩa. Dữ liệu được sắp xếp nghiêm ngặt theo timestamp và nén giá trị bằng thuật toán delta-of-delta cùng Gorilla float encoding. Trong thử nghiệm của chúng tôi, dung lượng lưu trữ trên đĩa thực tế giảm tới 72% so với Postgres. Dữ liệu ghi được append tuần tự vào WAL, loại bỏ hoàn toàn thao tác seek ngẫu nhiên của B-Tree. Khi dữ liệu hết hạn, InfluxDB xóa trọn vẹn các file shard bất biến trên filesystem chỉ trong vài mili-giây—hoàn toàn không cần chạy vacuum.

Từng bước triển khai: Chạy InfluxDB v2 trên Docker

Đóng gói InfluxDB với Docker Compose giúp bạn dựng một kho lưu trữ số liệu độc lập, dễ dàng tái lập chỉ trong vài phút. Hãy cùng bắt đầu từng bước cài đặt.

Bước 1: Thiết lập thư mục và thông tin xác thực

Chuẩn bị thư mục làm việc cho các tệp cấu hình của bạn:

mkdir -p ~/influxdb-iot && cd ~/influxdb-iot
touch docker-compose.yml .env

Thêm thông tin cấu hình ban đầu vào tệp .env. Hãy đảm bảo bạn tạo một token quản trị đủ mạnh:

DOCKER_INFLUXDB_INIT_USERNAME=iot_admin
DOCKER_INFLUXDB_INIT_PASSWORD=UseAStrongProdPassphrase2024!
DOCKER_INFLUXDB_INIT_ORG=plant_ops
DOCKER_INFLUXDB_INIT_BUCKET=telemetry_raw
DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=48f7b5a1b3924c738e4a9e403d15ef23a886f4a7c0d5

Khai báo service bên trong tệp docker-compose.yml:

services:
  influxdb:
    image: influxdb:2.7-alpine
    container_name: influxdb_iot
    restart: unless-stopped
    ports:
      - "8086:8086"
    environment:
      - DOCKER_INFLUXDB_INIT_MODE=setup
      - DOCKER_INFLUXDB_INIT_USERNAME=${DOCKER_INFLUXDB_INIT_USERNAME}
      - DOCKER_INFLUXDB_INIT_PASSWORD=${DOCKER_INFLUXDB_INIT_PASSWORD}
      - DOCKER_INFLUXDB_INIT_ORG=${DOCKER_INFLUXDB_INIT_ORG}
      - DOCKER_INFLUXDB_INIT_BUCKET=${DOCKER_INFLUXDB_INIT_BUCKET}
      - DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=${DOCKER_INFLUXDB_INIT_ADMIN_TOKEN}
      - DOCKER_INFLUXDB_INIT_RETENTION=30d
    volumes:
      - influxdb_data:/var/lib/influxdb2
      - influxdb_config:/etc/influxdb2

volumes:
  influxdb_data:
  influxdb_config:

Bước 2: Khởi chạy và kiểm tra

Khởi chạy container ở chế độ chạy nền (detached mode):

docker compose up -d

Theo dõi log của container để chắc chắn quá trình bootstrap tự động đã hoàn tất thành công:

docker logs -f influxdb_iot

Tìm dòng log có nội dung msg=Listening log_id=... service=tcp-listener transport=http addr=:8086. Lúc này, bucket telemetry_raw của bạn đã sẵn sàng hoạt động với chính sách lưu trữ cố định trong 30 ngày.

Bước 3: Đẩy dữ liệu Telemetry qua Line Protocol

Line Protocol gửi các chuỗi ký tự phân tách ngắn gọn trực tiếp qua HTTP. Bạn tránh được hoàn toàn overhead từ việc serialize JSON. Gửi thử một bản ghi mẫu từ thiết bị cảm biến số 12:

curl -i -XPOST "http://localhost:8086/api/v2/write?org=plant_ops&bucket=telemetry_raw&precision=s" \
  --header "Authorization: Token 48f7b5a1b3924c738e4a9e403d15ef23a886f4a7c0d5" \
  --data-raw "machine_metrics,device_id=sensor_12,line=assembly_4 temperature=42.3,vibration=0.018,kw_load=3.75"

Cấu trúc của Line Protocol được chia thành ba phần rõ rệt:

  • Measurement (machine_metrics): Đóng vai trò là nhóm dữ liệu đích (tương tự như một bảng trong SQL).
  • Tag Set (device_id=sensor_12,line=assembly_4): Các cặp metadata dạng chuỗi được TSM engine đánh chỉ mục, giúp lọc dữ liệu với tốc độ microsecond.
  • Field Set (temperature=42.3,vibration=0.018,kw_load=3.75): Các giá trị payload có định kiểu và không đánh chỉ mục, chứa số liệu đo thực tế.

Bước 4: Truy vấn và Downsample bằng Python

Trong các hệ thống giám sát production, các số liệu này được kéo về downstream để đánh giá cảnh báo và hiển thị lên dashboard. Cài đặt thư viện client chính thức của InfluxDB:

pip install influxdb-client

Đoạn mã bên dưới truy xuất dữ liệu nhiệt độ trong 2 giờ qua và tính giá trị trung bình trượt mỗi 5 phút bằng Flux:

from influxdb_client import InfluxDBClient

URL = "http://localhost:8086"
TOKEN = "48f7b5a1b3924c738e4a9e403d15ef23a886f4a7c0d5"
ORG = "plant_ops"
BUCKET = "telemetry_raw"

client = InfluxDBClient(url=URL, token=TOKEN, org=ORG)
query_api = client.query_api()

flux_query = f'''
from(bucket: "{BUCKET}")
  |> range(start: -2h)
  |> filter(fn: (r) => r._measurement == "machine_metrics")
  |> filter(fn: (r) => r._field == "temperature")
  |> aggregateWindow(every: 5m, fn: mean, createEmpty: false)
  |> yield(name: "rolling_mean")
'''

tables = query_api.query(org=ORG, query=flux_query)

for table in tables:
    for record in table.records:
        ts = record.get_time().strftime("%Y-%m-%d %H:%M:%S")
        device = record.values.get("device_id")
        val = record.get_value()
        print(f"[{ts}] {device} | Nhiệt độ TB: {val:.2f}°C")

client.close()

Kết quả vận hành

Việc tách riêng các luồng ghi cảm biến tần suất cao sang InfluxDB đã giải tỏa áp lực ngay lập tức cho cơ sở dữ liệu quan hệ chính của chúng tôi. Disk I/O giảm mạnh từ 98% xuống chỉ còn 12%. Nhu cầu lưu trữ giảm từ 18 GB mỗi tuần xuống dưới 5 GB. Quan trọng nhất, các truy vấn tổng hợp dữ liệu 30 ngày trước đây thường xuyên bị timeout thì nay hoàn thành ổn định trong chưa đầy 65 mili-giây.

Share:
Tags: