定番の「MQTT + AWS IoT Core + Timestream for InfluxDB + Grafana」を、実運用目線でひと工夫 ― hot/coldデータの組み合わせと、スケール可能なデバイスモックで仮想1000台×30日分を検証する

1 はじめに

製造ビジネステクノロジー部の平内（SIN）です。

IoT のデータ可視化といえば、「MQTT + AWS IoT Core + 時系列データベース + Grafana」が定番の構成かと思います。そこで今回は、この定番構成を Amazon Timestream for InfluxDB を軸に、改めて一通り確認してみました。

検証にあたって以下の点を考慮してみました。

• コールドパス（過去）とホットパス（直近）のシームレスな表示 … 長期データは S3 Tables（Iceberg）+ Athena、直近データは Timestream for InfluxDB に分け、Grafana の1枚のパネルに継ぎ目なく重ねます
• デバイスモックを Fargate で実装（数万台規模の試験にも対応） … 実機ゼロで仮想デバイス1000台を実際に MQTT 接続します。タスク数・台数はパラメータ化してあるので、数万台規模の負荷試験にも簡単に拡張できるようにしました
• 過去データは「貯める」のではなく「生成」した … 待ち時間ゼロで、過去30日分（約4,300万行）を直接 Iceberg にバックフィルします。日周・週次・個体差・ノイズに加えて、4種類の異常イベントを仕込み、現実味のある波形にしています
• IoT Core から Kinesis 経由のバッチ処理でInfluxDB へ接続 … Basic Ingest + 500件バッチで Lambda 実行回数を 1/500 にしました
• 長期データは Amazon S3 Tables（マネージドな Iceberg テーブル）に置いて高速にアクセス … 生成した過去30日分は、S3 Tables 上の Iceberg テーブルに保存します。データ本体は列指向で圧縮された Parquet、それを束ねる メタ情報（マニフェスト） の管理や、小さなファイルの自動整理（コンパクション）を AWS に任せられ、Athena からそのまま SQL で集計できます。時系列の長期データを InfluxDB 側に抱えさせずに済むため、コスト面でも扱いやすくなります（パーティションの効きは第5章で実測します）
• S3 Tables の権限設計の整理 … テーブルを作るだけでは Athena から読めず、必要な権限が Athena / Glue / Lake Formation / S3 / S3 Tables の5つのサービスにまたがります。Grafana 用に最小権限の専用ユーザーを用意するにあたり、この5層をどう組むかを整理しました

コードは以下に置きました。

https://github.com/furuya02/iot-timestream-influxdb-grafana-1000-devices

2 全体アーキテクチャ

(1) ホット/コールドの2パス構成

可視化はローカル Docker の Grafana で行います。Athena データソース（コールド）と InfluxDB データソース（ホット）を Mixed データソースとして、1枚のパネルに重ねます。

(2) 2パスに分けた理由とデータストア選定

先に結論を述べると、双方の得意分野だけを使いたいからです。Athena は1クエリ数秒かかるため自動更新は分単位が限度で、一方の InfluxDB は秒単位に追従できますが、長期保持はインスタンスのストレージコストが乗ってきます。そこで「直近は InfluxDB、長期は S3 Tables（Iceberg）」と役割を分けました。

時系列データベースのマネージドサービスは、Amazon Timestream for LiveAnalytics が2025年6月20日以降は新規顧客の利用不可となり（Amazon Timestream FAQs）、現行は後継の Amazon Timestream for InfluxDB が選択肢です。ただしインスタンス時間課金（料金）のため、今回はホットパス専用の最小インスタンス（db.influx.medium / Single-AZ）とし、長期は S3 Tables（Iceberg）に逃がしました。1000台 × 1Hz（約1000 points/秒）と書き込み負荷が小さく、長期データも Iceberg 側に持たせているため、このサイズで足ります。Parquet + Snappy 圧縮なら30日・4,300万行が数 GB に収まります。

3 CDK

インフラは全て AWS CDK（TypeScript）で構築します。時間課金となる Timestream for InfluxDB（InfluxStack）だけは独立したスタックに分離し、ほぼ無料のスタックを先にデプロイして準備を整えてから、InfluxStack は計測の直前にデプロイします。この分離は消し忘れ対策の要でもあり、その狙いは第9章で述べます。

デプロイ手順は以下のとおりです。

git clone https://github.com/furuya02/iot-timestream-influxdb-grafana-1000-devices.git
cd iot-timestream-influxdb-grafana-1000-devices/cdk
pnpm install
pnpm exec cdk bootstrap                # 初回のみ

# 1. 無料ゾーンを先にデプロイ
pnpm exec cdk deploy ColdStack PlumbingStack LoadGenStack

# 2. 過去30日分を生成して Iceberg へ投入（コールドパスの準備）
cd ../backfill && python3 -m pip install -r requirements.txt
export WAREHOUSE_ARN=$(aws cloudformation describe-stacks --stack-name ColdStack \
  --query "Stacks[0].Outputs[?OutputKey=='TableBucketArn'].OutputValue" --output text)
python3 create_table.py && python3 generate.py

# 3. デバイス証明書を発行して S3 へ配布
cd ../scripts && bash issue-cert.sh

# 4. ここから時間課金。InfluxDB をデプロイし、トークンを自動配線
cd ../cdk && pnpm exec cdk deploy InfluxStack
bash ../scripts/wire-influx-token.sh

# 5. 仮想デバイス1000台を起動（5タスク × 200台）
bash ../scripts/run-devices.sh 1000 5

# 6. 計測が終わったら片付け（InfluxDB を最優先に削除）
bash ../scripts/stop-devices.sh
pnpm exec cdk destroy InfluxStack          # 時間課金を最優先で停止
bash ../scripts/destroy-all.sh             # cdk destroy を妨げる要素を先に掃除（下記）
pnpm exec cdk destroy --all                # 残りの全スタックを削除

なお、台数・タスク数・Kinesis のシャード数・InfluxDB のインスタンスタイプは、CDK の context でパラメータ化しています（例: -c devices=30000 -c shards=32）。

片付けで1点、注意があります。CDK 管理外のリソースを先に消しておかないと、cdk destroy が失敗します。具体的には、デバイス証明書が IoT ポリシーに紐付いていると PlumbingStack を削除できず、pyiceberg で作った Iceberg テーブルが残っていると ColdStack の S3 Tables バケットを削除できません。上記の destroy-all.sh は、この2つ（証明書・S3 Tables のテーブル）と、Grafana 用の IAM ユーザーを先に掃除します。

4 過去30日を「生成」する

(1) 波形の設計

過去データを実際に30日間貯めるのは、待ち時間が長すぎます。そこで、以下の式で「それらしい」波形を直接生成することにしました。

value(t) = base(device)                     # 個体ごとの基準値
         + 日周成分（sin 24h、週末は振幅 0.6 倍）
         + 個体差（デバイス固定のオフセット・ゲイン）
         + ガウシアンノイズ
         + 異常イベント（下表）

1000台 × 1サンプル/分 × 30日 = 4,320万行になります。実測では、仕込んだ欠測イベントの分だけ少ない 43,184,280行 が投入され、欠測数（15,720行）が設計値と完全に一致しました。生成ロジックが意図どおり動いていることが確認できました。

Github generate.py

(2) 仕込んだ4種類の異常

後半の章で Athena の SQL で検出するために、あらかじめ4種類の異常イベントを仕込んでおきます。

(3) タイムゾーンの扱い

生成・投入・ライブ送信のすべてを UTC で統一し、Grafana の表示だけ JST にしています。ここを混在させると、ホットとコールドの継ぎ目がズレてしまうため、実装前に統一しておく必要があります。

5 S3 Tables を Athena から読むための権限設計

S3 Tables を Athena（および Grafana）から読むには、複数のサービスにまたがる権限を整理しておく必要があります。ここでは、その全体像をまとめます。

(1) テーブル作成と統合の有効化

S3 Tables のテーブルバケットは CDK（CfnTableBucket）で作成し、名前空間・テーブル定義は pyiceberg で行います。ここで一点、押さえておく必要があります。テーブルを作成しただけでは、Athena からは参照できません。アカウント×リージョン単位で「S3 Tables と AWS 分析サービスの統合」を一度有効化しておきます（S3 Tables ドキュメント）。有効化すると、Glue に s3tablescatalog というフェデレーテッドカタログが作られます。

なお、このカタログは Athena のデータソース選択のドロップダウンには現れないため、SQL の中でフル修飾して参照します。

SELECT count(*) FROM "s3tablescatalog/iot-demo-20260808"."iot_demo"."telemetry";

(2) 専用IAMユーザーに必要な5層の権限

Grafana からは、最小権限の読み取り専用 IAM ユーザーで接続します。S3 Tables を Athena 経由で読む場合、必要な権限は次の5つのサービスにまたがります。

整理のうえで押さえておきたい点が2つあります。

• 第4層のとおり、テーブル本体への s3:GetObject は不要です。データ本体は第3層の Lake Formation が一時クレデンシャルを発行して読ませるため、S3 権限は「Athena 結果バケットだけ」で足ります
• 第5層の s3tables:* が不足していると、クエリは CATALOG_NOT_FOUND: Catalog '...' does not exist を返します。Lake Formation は権限のないリソースを not found として扱うため、エラー文言（カタログが無い）と実際の原因（s3tables:* の不足）が一致しない点に注意が必要です

Github iam-policy.json

(3) パーティションの効きとコスト実測

パーティションは days(event_time) + site としました。Iceberg のメタデータテーブル $partitions を見ると、event_day × site ごとにファイルが分かれていることを確認できます（各パーティションは 180,000 行 = 125台 × 1440分）。

SELECT * FROM "s3tablescatalog/iot-demo-20260808"."iot_demo"."telemetry$partitions" LIMIT 20;

このパーティションが、スキャン量に効いてきます。拠点別のフルスキャン集計では、388.11 MB / 約2.5秒（Athena 料金換算で約0.2セント）をスキャンしました。

一方、1日×1拠点に絞ると、パーティション刈り込みが効いて 1.62 MB まで減ります（約1/240）。

また、Iceberg の count(*) はマニフェスト（メタ情報）だけで応答するため、スキャン量ゼロ（課金ゼロ）で行数を確認できます。実行結果の「スキャンしたデータ」がハイフン（-）になっている点が特徴です。

6 ホットパス：Fargate の仮想1000台から InfluxDB まで

(1) IoT Core → InfluxDB のネイティブ経路が無い

IoT ルールアクションの一覧（ドキュメント）を見ると、Timestream（LiveAnalytics）向けアクションはあるものの、Timestream for InfluxDB 向けはありません。そこで、間に処理を挟むことになります。検討した3案は以下のとおりです。

あわせて、デバイスは通常のトピックではなく $aws/rules/iot_demo_ingest に直接 publish する Basic Ingest を使い、メッセージング料金を抑えています。

(2) Lambda で line protocol に変換して書き込む

Lambda は、Kinesis のバッチを InfluxDB の line protocol にまとめて変換し、v2 互換の write API に1回で書き込みます。実測でも1バッチ数百行を十数 ms で処理できており、変換自体はごく軽い処理でした。

# handler.py（要点抜粋）
f"telemetry,device_id={dev},site={site},firmware={fw} "
f"temperature={temp},humidity={hum},vibration={vib},"
f'ingest_time={ingest_ms}i,source="live" {ts_ns}'

Github handler.py

なお、この経路で2点、注意しておく点があります。

• VPC 到達性 … Lambda を VPC 内（NAT なし）に置くと、InfluxDB のパブリックエンドポイントに到達できず、write がタイムアウトします。検証では Lambda を非 VPC とし、InfluxDB 側のセキュリティグループをトークン認証前提で開放する割り切りにしました。ただし、これはあくまで検証用です。本番では Lambda・InfluxDB とも VPC 内に置き、SG を厳格化するのが原則かと思います
• write API の org パラメータ … ?bucket=...&precision=ns だけでは HTTP 400（Please provide either orgID or org）になります。?org=iot&bucket=iot_demo&precision=ns と、org の指定が必須でした

加えて、Timestream for InfluxDB の作成時パラメータにも制約があります。管理者パスワードは英数字のみ（[a-zA-Z0-9]+）で、記号を含むパスワードは受け付けられません。

(3) Fargate で仮想デバイス1000台

シミュレータは Python（asyncio + aiomqtt）で、1プロセス内で多数の MQTT 接続を非同期に回します。Fargate 5タスク × 200台 = 1000台で運用し、タスク数・台数はスクリプトの引数で変更できます。

bash scripts/run-devices.sh 1000 5    # 5タスク × 200台
bash scripts/run-devices.sh 30000 30  # 引数を増やせばそのまま拡張（30タスク × 1000台）

Github simulate.py

設計上のポイントは、以下の2点です。

• 証明書は1枚を全タスクで共有し、IoT ポリシーで ClientId を sim-* に制約、Publish 先を Basic Ingest トピックのみに制限しています。なお、本番では1デバイス1証明書 + ポリシー変数が原則かと思います。証明書はコンテナイメージに焼き込まず、起動時に S3 から取得するようにしました
• ライブの値も過去データと同じ波形（同じシード・同じ式）で生成しています。これが、次章の「継ぎ目のない表示」の仕掛けになります

接続レートについても補足します。事前の想定では「1000台の一斉接続で IoT Core の接続レート上限（アカウントあたり概ね500/秒、Service Quotas）に達し、スロットリングが再現できる」と考えていました。ところが実測では、各タスクの TLS ハンドシェイク自体が律速となり（1000接続を5タスクに分散しているため、1タスクあたり約200接続）、接続レート上限には達せず、スロットリングは発生しませんでした。「1000台一斉＝必ず詰まる」わけではなく、壁は接続を張る速さの側にある、というのが実測結果です。

7 Grafana で過去と直近を継ぎ目なく表示する

(1) データソースの設定

Grafana はローカル Docker で起動し、コールド（Athena）とホット（InfluxDB）の2つのデータソースを、いずれもプロビジョニング（YAML）で設定します。認証情報（Athena 用 IAM ユーザーのアクセスキー、InfluxDB のエンドポイントとトークン）は環境変数から注入し、YAML ファイル自体には秘密情報を書きません。コンテナ起動時に、両データソースが自動で登録されます。

• Amazon Athena データソース（コールドパス）… カタログ（AwsDataCatalog）・データベース・ワークグループ・結果出力先を YAML で固定します。公式プラグインを起動時に導入します
• InfluxDB データソース（ホットパス）… Grafana に標準搭載されています。Query language は Flux、Organization は iot を指定します

Github athena.yaml / influxdb.yaml

(2) コールドパスのダッシュボード

まずは過去30日の生成データだけで、4枚のパネルを作ってみました。拠点別の日周のうねり、sim-0042 のドリフトの立ち上がり、site-c のアクティブ台数、温度スパイクが一望できます。

(3) 継ぎ目パネル

本記事の主題である、過去と直近を1枚に重ねるパネルです。1枚のパネルで2つの異なるデータソース（Athena と InfluxDB）を同時に使うには、パネルの データソースに「-- Mixed --」を選択します。すると、クエリごとにデータソースを個別指定できるようになります。

設定手順は次のとおりです。

1. パネルを新規作成し、Data source で「-- Mixed --」を選ぶ
2. クエリA のデータソースを Athena にして、過去データの集計 SQL を書く（Format data frames as は Time Series）
3. ＋ Add query でクエリBを足し、データソースを InfluxDB にして、直近ライブの Flux を書く

クエリA（Athena）— S3 Tables のカタログをフル修飾し、$__timeFilter でパネルの時間範囲に連動させます。

SELECT date_trunc('minute', event_time) AS time, avg(temperature) AS "past(Athena)"
FROM "s3tablescatalog/iot-demo-20260808"."iot_demo"."telemetry"
WHERE $__timeFilter(event_time)
GROUP BY 1 ORDER BY 1

クエリB（InfluxDB / Flux）— v.timeRangeStart / v.timeRangeStop でパネルの時間範囲に、v.windowPeriod で表示幅に応じた集計粒度に、それぞれ自動追従します。

from(bucket: "iot_demo")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r._measurement == "telemetry" and r._field == "temperature")
  |> group()
  |> aggregateWindow(every: v.windowPeriod, fn: mean, createEmpty: false)

両クエリともパネルの時間範囲に追従するため、時間レンジを変えるだけで見え方が変わります。「Last 2 days」では過去2日の波形の右端に直近のライブが連続し、「Last 3 hours」までズームインすると、過去（Athena）と今まさに Fargate の1000台が送っているライブ（InfluxDB）が重なる継ぎ目が見えます。過去は生成データ、直近は実際のライブという2つの世界が、同じ波形・同じ時間軸で1枚に繋がります。

8 仕込んだ異常の検出について

第4章で仕込んだ4種類の異常（ドリフト・完全欠測・スパイク・一斉オフライン）は、いずれも Athena の SQL で検出できます。サンプルの検出SQLをリポジトリに置きましたので、興味があれば参照してください。

Github anomaly_detection.sql

9 コストと、消し忘れ対策

(1) 実験1回あたりの概算

検証全体を通しても、数千円のオーダーに収まる見込みです。ポイントは、コストの大半が「InfluxDB インスタンスの稼働時間」で決まる、という点です。

(2) 消し忘れを構造で防ぐ

この構成で唯一「放置すると高額」になるのが、Timestream for InfluxDB です。時間課金のため、放置すると月数万円のオーダーになり得ます。そこで今回は、以下で構造的に対策してみました。

• CDK をスタック分割し、InfluxDB（InfluxStack）だけを独立させています。計測が終わったら cdk destroy InfluxStack の1コマンドで課金を停止できます
• データ本体は S3（Iceberg）にあるため、InfluxDB を消しても過去データは失われません。「消して良いものを消しやすく作る」のが、ホット/コールド分離のもう一つの利点かと思います
• AWS Budgets の閾値アラート（50% / 100%）を、CDK で最初に作成しています

10 まとめ

定番の「MQTT + AWS IoT Core + Timestream for InfluxDB + Grafana」構成を、仮想デバイス1000台と生成データ30日分で一通り確認してみました。今回の検証で分かったことを、5点にまとめます。

• 過去データは「生成」すれば、実機も長期運用も待たずに、その日のうちに30日分のダッシュボードが手に入ります
• ホット（InfluxDB）とコールド（Iceberg + Athena）を分け、ライブ側も同じ波形で送ることで、過去と直近を1枚のパネルに継ぎ目なく表示できます
• IoT Core から InfluxDB への直接経路は無いため、Basic Ingest + Kinesis + バッチ Lambda が現実的な選択になります
• S3 Tables を専用ユーザーで扱う際は、Athena / Glue / Lake Formation / S3 / s3tables:* の5層の権限を整理する必要があります。特に、Lake Formation が権限不足を「not found」に見せる挙動は、原因の特定を分かりにくくするため注意です
• デバイスモックを Fargate + パラメータ化しておけば、同じ仕組みのまま数万台の負荷試験にも進めそうです

「定番構成をこれから触ってみる」ときの、たたき台になれば嬉しいです。

コード一式（CDK・生成スクリプト・シミュレータ・SQL）は、以下に置きました。

https://github.com/furuya02/iot-timestream-influxdb-grafana-1000-devices

11 参考リンク

• Amazon Timestream for InfluxDB 開発者ガイド
• Amazon Timestream FAQs
• Amazon Timestream 料金
• AWS IoT ルールアクション
• AWS IoT Basic Ingest
• AWS IoT Core Service Quotas
• Amazon S3 Tables
• Amazon Athena 料金
• Grafana Amazon Athena data source