実際の課題：センサーストリームで悲鳴を上げるリレーショナルデータベース
昨年、私はある製造業のクライアント向けに、500台の産業用IoT振動・温度プローブを対象としたデータ取り込みパイプラインのアーキテクチャ設計を支援しました。各機器は温度、3軸振動、消費電力を2秒ごとに送信します。これは毎秒250件のペイロードに相当し、24時間で約2,160万行ものデータがディスクに書き込まれる計算になります。
当初、私たちはこのストリームを8コア・32GB RAMの仮想マシン上でチューニングしたPostgreSQL 14インスタンスへ直接流し込んでいました。最初の10日間はパフォーマンスも安定していました。しかし3週目に入る頃には、30日間の移動平均を集計するクエリが行き詰まり、60秒でタイムアウトし始めたのです。
ディスクI/Oの飽和度は98%に達しました。テーブルの複合B-Treeインデックスは、利用可能なRAM容量を大幅に超える45GB以上にまで膨れ上がりました。バッファキャッシュのミスヒットが急増し、受信した書き込みデータはRedisに滞留、最終的にはデータ取り込みワーカーがメモリ枯渇によってダウンしてしまいました。
Postgresは極めて優れたエンジンです。MySQLやMongoDBも同様です。しかし、継続的かつ追記専用（append-only）のセンサーテレメトリを汎用リレーショナルデータベースに流し込むと、インデックスのチューニング程度では到底解決できない構造的なディスクボトルネックが発生します。
根本原因の分析：なぜ従来のエンジンはストリーム書き込みで破綻するのか
根本的な問題は、汎用エンジンがディスク上でデータをどのように配置・管理するかにあります。
- インデックス維持によるスラッシング：新しいレコードが挿入されるたびに、テーブルのインデックスが更新されます。テーブルやインデックスのサイズがRAMを超えると、ストレージ層はディスク上に分散したB-Treeページの読み取りと更新に追われるようになります。その結果、スループットは急激に低下します。
- 行指向によるオーバーヘッドの増大：リレーショナルシステムは行タプル全体を連続して書き込みます。工場内センサーの直近6時間の平均温度を算出するだけでも、エンジンはディスクから無関係な振動、消費電力、ステータスなどのカラムを大量にスキャンしなければなりません。このデータ構造は、メモリとバスに莫大な負荷をもたらします。
- 過度なデータ削除（パージ）負荷：90日の保持期間を過ぎたセンサーレコードを削除する場合、通常はバッチ形式で
DELETE文を実行します。しかしPostgresやMySQLでは、5,000万行を削除すると大量の不要タプル（デッドタプル）が残ります。これがテーブルの断片化を引き起こし、WAL（Write-Ahead Log）を肥大化させ、積極的なオートバキュームの実行を誘発して書き込み処理を圧迫します。
アーキテクチャの比較：汎用データベース vs 時系列特化型エンジン
メトリクスストリームを扱うチームは、通常次の3つの選択肢を検討します。
1. リレーショナルデータベース（PostgreSQL / TimescaleDB）
標準のPostgresは強固なACID保証とリレーショナル結合を提供しますが、素のテーブル構造のままでは時間による範囲パーティショニングなどの徹底した手動パーティショニングが不可欠です。TimescaleDBはハイパーテーブル（hypertables）によってこれらの問題の多くを解消しますが、書き込みバッファやインデックス構造のために依然として小さくないメモリ消費が発生します。
2. ドキュメント指向データベース（MongoDB）
MongoDBは書き込みを水平スケールでき、事前のマイグレーションなしに疎データや多様なペイロードを柔軟に処理できます。しかし、時間軸でのダウンサンプリングを実行するには独自の集計パイプラインやバックグラウンドワーカーが必要です。これらの処理は、CPUやWiredTigerキャッシュの領域を巡ってデータ取り込みスレッドと直接競合することになります。
3. 時系列特化型エンジン（InfluxDB）
InfluxDBは、カラムナ（列指向）ディスクストレージと組み合わせたTime-Structured Merge Tree（TSM）エンジンを採用しています。データをタイムスタンプ順に厳密に並べ替え、delta-of-deltaやGorilla浮動小数点エンコーディングを用いて値を圧縮します。私たちの検証では、ディスク占有容量はPostgresと比較して72%削減されました。書き込みはWALにシーケンシャルに追加されるため、ランダムなB-Treeシークは完全に回避されます。また、保持期間を過ぎたデータはイミュータブルなファイルシャードごとミリ秒単位でファイルシステムから破棄されるため、バキューム処理などのメンテナンスは一切不要です。
ステップ・バイ・ステップ：DockerでInfluxDB v2を動かす
Docker Composeを使ってInfluxDBをコンテナ化することで、分離された再現性の高いメトリクスストアを数分で立ち上げることができます。具体的なセットアップ手順を見ていきましょう。
ステップ1：ディレクトリと認証情報の設定
設定ファイル用の作業ディレクトリを準備します。
mkdir -p ~/influxdb-iot && cd ~/influxdb-iot
touch docker-compose.yml .env
.envに初期設定用の認証情報を追加します。管理者トークンには必ず強力な値を生成して設定してください。
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
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:
ステップ2：起動と動作確認
バックグラウンドでコンテナを起動します。
docker compose up -d
コンテナのログをストリーミング表示し、初期セットアップが自動完了したことを確認します。
docker logs -f influxdb_iot
msg=Listening log_id=... service=tcp-listener transport=http addr=:8086というログ行を確認してください。これで、30日間の保持期間が設定されたtelemetry_rawバケットが利用可能な状態になりました。
ステップ3：Line Protocolによるテレメトリの送信
Line Protocolは、区切り文字を用いたコンパクトな文字列をHTTP経由で直接送信するプロトコルです。JSONシリアライズのオーバーヘッドを完全に排除できます。センサーユニット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"
Line Protocolの構造は、大きく3つのセクションに分かれています。
- Measurement（
machine_metrics）：対象のグルーピング（SQLのテーブルに相当）として機能します。 - Tag Set（
device_id=sensor_12,line=assembly_4）：マイクロ秒単位の高速フィルタリングのためにTSMエンジンによってインデックス化される文字列メタデータです。 - Field Set（
temperature=42.3,vibration=0.018,kw_load=3.75）：インデックス化されない型付きのペイロードで、実際の測定値が格納されます。
ステップ4：Pythonによるクエリとダウンサンプリング
本番のモニタリングパイプラインでは、これらのメトリクスを下流で取得してアラートの判定やダッシュボードの構築を行います。公式のInfluxDBクライアントをインストールします。
pip install influxdb-client
以下のスクリプトは、Fluxを使用して過去2時間の温度メトリクスを取得し、5分ごとの移動平均を計算します。
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} | 平均温度: {val:.2f}°C")
client.close()
運用の成果
高頻度なセンサー書き込み処理をInfluxDBに分離したことで、プライマリのリレーショナルデータベースの負荷は劇的に軽減されました。ディスクI/Oは98%から12%へと低下し、ストレージ消費量も週あたり18GBから5GB未満へと縮小しました。そして何よりも、以前はタイムアウトしていた30日分の集計クエリが、常に65ミリ秒未満で完了するようになったのです。
