概要
リアルタイムデータの蓄積と加工は、ログ分析、モニタリング、機械学習前処理など、多くのユースケースで重要な要素です。
本記事では、Amazon Kinesis Data Firehose と AWS Lambda を用いて、次のような構成を実現する方法を紹介します：
- リアルタイムでストリーミングデータを取り込み
- 必要なデータ加工処理をLambdaで自動化
- 処理済みデータをS3に永続化
このアーキテクチャは、完全サーバレスでスケーラブルな設計が可能であり、初期構築や運用の手間を最小限に抑えることができます。
システム構成イメージ
データ処理の流れ
- データ送信元 は、Firehose の Delivery Stream に対して HTTP API などを用いてデータを送信します。
- Firehose は、受信したデータをリアルタイムで Lambda 関数 に渡します。
- Lambda 関数 内では、データ整形・バリデーション・フィールドの追加や変換など、自由に加工処理を実装可能です。
- 加工が完了したデータは、Amazon S3 に自動で格納され、後続のバッチ処理やAthenaなどによるクエリ分析に活用できます。
Lambda 変換関数のサンプル（Python）
import json
import base64
def lambda_handler(event, context):
output = []
for record in event['records']:
# Base64デコードしてJSONとしてパース
payload = json.loads(base64.b64decode(record['data']))
# 加工処理：例としてタイムスタンプの追加
payload['processed_at'] = context.aws_request_id
# 加工後のデータを再エンコード
encoded_data = base64.b64encode(json.dumps(payload).encode('utf-8')).decode('utf-8')
output_record = {
'recordId': record['recordId'],
'result': 'Ok',
'data': encoded_data
}
output.append(output_record)
return {'records': output}
Firehose の主な設定項目
| 設定項目 | 内容 |
|---|---|
| Delivery Stream Type | Direct PUT または Kinesis Stream 経由を選択可能 |
| Lambda 関数指定 | 加工処理用のLambda関数を指定（リアルタイム変換を有効化） |
| S3 設定 | 保存先バケット、キーPrefix、圧縮（GZIPなど）の指定が可能 |
| ログ設定 | CloudWatch Logs にFirehose処理結果を出力してデバッグや監視が可能 |
なぜこの構成が有用なのか？
| 特徴 | 説明 |
|---|---|
| 完全サーバレス | EC2やECSを使用せず、スケーラブルで保守不要なデータパイプラインが構築可能 |
| 拡張性 | Lambdaのロジック変更で柔軟にデータ加工処理をアップデート可能 |
| コスト効率 | バッチ転送＆圧縮でストレージコストや転送コストを最小限に |
| 保守性 | マネージドサービス活用により、スケーリング・パッチ・稼働監視が不要 |
まとめ
| ユースケース | 説明 |
|---|---|
| Webログの整形と分析保存 | JSON形式に変換し、AthenaやQuickSightでのクエリ分析が可能に |
| IoTセンサーのデータ拡張と蓄積 | 不要フィールド除去やメタ情報追加などをLambdaで前処理 |
| ユーザー操作ログのフィルタ保存 | 特定イベントや操作のみを抽出して保存し、後続分析を効率化 |
Lambdaによるデータ前処理をFirehose内で実現することで、後続のバッチ処理やBI分析を意識したデータ構造をS3に直接格納できます。拡張性・保守性・運用効率の三拍子が揃ったサーバレスアーキテクチャです。