Data Firehose本番運用ガイド｜Iceberg/Snowflake宛先実践

1. この記事について

1-1. 本記事のゴール

Amazon Data Firehose という名前は聞いたことがあっても、「結局 S3 に置くだけの中継サービス」という理解で止まっていないでしょうか。

本記事は、Amazon Data Firehose を本番運用の観点から掘り下げ、配信ストリームの設計原則から、複数ある宛先の選定基準、そして2024年に追加された Apache Iceberg テーブル・Snowflake という新宛先の実践的な位置づけまでを一気通貫で整理することをゴールとしています。単なる機能一覧の紹介ではなく、「なぜその設計を選ぶのか」という判断軸を提供することに主眼を置きます。

本記事を読み終えたとき、読者の方には次の4点に到達していただくことを目指します。

• 配信ストリームのソース(Direct PUT / Kinesis Data Streams / Amazon MSK)ごとの特性と、バッファリング設定(サイズ・インターバル)が配信レイテンシとコストにどう影響するかを説明できる状態
• Amazon S3 / Amazon Redshift / OpenSearch Service / Splunk という既存宛先と、Apache Iceberg テーブル・Snowflake という新宛先を、ユースケースに応じて使い分けられる状態
• Lambda によるデータ変換(動的パーティショニング・フォーマット変換)の制約(ペイロード上限・タイムアウト)を踏まえて設計できる状態
• 2024-02 の名称変更(Kinesis Data Firehose → Amazon Data Firehose)が、表示名のみの変更であり API・CLI・IAM・Terraform 上の識別子には及んでいないという非対称性を正確に把握している状態

成果物イメージとしては、配信ストリームを Direct PUT または Kinesis Data Streams から起動し、動的パーティショニングを設定した上で、S3(またはIceberg テーブル)への配信が本番運用に耐える設定(バッファ・リトライ・監視)で稼働している状態を思い浮かべてください。

これらの到達点は、単なる機能紹介にとどまりません。本サービスを新規に「導入するかどうか」を判断するための材料と、既に導入済みの配信ストリームを「どう本番運用の水準まで引き上げるか」という改善の両方に活用できることを狙っています。特に、複数の宛先を段階的に追加してきた結果、設定が属人化してしまっている配信ストリームの棚卸しにも、本記事の整理軸をそのまま応用できます。

なお、本記事が扱わない範囲もあらかじめ明確にしておきます。動的パーティショニングや Lambda 変換そのものの基礎的な仕組み、CloudWatch Logs や VPC Flow Logs をソースとした具体的なユースケースは、本記事のスコープ外とし、該当する既存記事に譲ります(詳細は1-3で整理します)。

本記事の章立ては次のとおりです。具体的な設定手順を急ぎ確認したい場合は、目的の章から読み進めることもできます。

すでに配信ストリームを本番運用しており、Iceberg / Snowflake宛先の採用のみを検討している場合は§5から、これから新規に配信ストリームを設計する場合は§2から順に読み進める構成になっています。

本記事が特に有効なのは、すでに Kinesis Data Streams や既存のログ収集基盤を運用しており、これから配信先を S3 データレイクや Redshift・OpenSearch といった分析基盤へ拡張しようとしている組織です。単一アカウント・単一リージョンの小規模な検証環境でも本記事の内容は適用できますが、複数の宛先を使い分けたり、Iceberg / Snowflake という新宛先の採用を本番導入レベルで検討していたりする組織ほど、本記事が提供する判断材料の価値は大きくなります。

1-2. 読者像

本記事が想定する読者は、すでに何らかの形でストリーミングデータ基盤に触れているデータ基盤エンジニア・SRE の方々です。具体的には、次のような課題感を持つ方を想定しています。

配信ストリームの設計判断に自信が持てない

Direct PUT・Kinesis Data Streams・Amazon MSK という複数のソース選択肢があり、どの組み合わせが自組織のユースケースに適しているか、公式ドキュメントを読んでもなかなか判断がつかない状況です。

宛先が増え続けて選定基準が曖昧になっている

S3・Redshift・OpenSearch・Splunk に加え、2024年には Iceberg テーブルと Snowflake が新宛先として加わりました。既存のデータレイク・DWH 構成との組み合わせで、どの宛先を選ぶべきか整理できていない状況です。

改称に伴う命名の混乱に対する不安

2024-02 に Kinesis Data Firehose から Amazon Data Firehose へと表示名が変わったものの、Terraform や IAM ポリシー、CLI コマンドのうち変更された部分と変更されていない部分を、正確に把握できていない状況です。

こうした課題は、たとえば次のようなシーンで顕在化します。

• 新規にストリーミングパイプラインを設計する際、Kinesis Data Streams と Firehose の Direct PUT のどちらを起点にすべきか判断に迷う場合
• 既存の S3 データレイクに Iceberg テーブルを導入したいが、Firehose からの直接書き込みが選択肢になり得るのか分からない場合
• Terraform で aws_kinesis_firehose_delivery_stream という旧名のリソース名を見て、改称が正しく反映されていないのではと不安になる場合
• 複数の宛先(S3・Redshift・OpenSearch等)を並行運用しており、Iceberg / Snowflake宛先への移行・追加を検討しているが、既存パイプラインへの影響範囲が見積もれない場合
• Lambda変換や動的パーティショニングを使ったパイプラインでエラーが散発しており、上限値のどれに抵触しているのか切り分けられない場合

導入検討の初期段階でセルフチェックしておきたい項目は次の3点です。

• 配信ストリームのソースとして、Direct PUT・Kinesis Data Streams・Amazon MSKのいずれを起点とするか、既存のデータ収集基盤との整合性を確認したか
• 宛先候補が複数ある場合、既存宛先(S3/Redshift/OpenSearch/Splunk)で要件を満たせるか、Iceberg/Snowflakeという新宛先でなければ実現できない要件があるかを切り分けたか
• Terraform・IAMポリシー・CLIスクリプトが、改称後もkinesis/firehose表記のままで問題ないという前提を、チーム内で共有できているか

前提知識としては、AWS アカウントでの基本的なリソース作成操作、ストリーミングデータの基礎概念(レコード・シャード・パーティション)、IAM ポリシーの読み書きができることを想定しています。Kinesis Data Streams や Amazon MSK 自体の詳細な設計・運用は前提としますが、本記事内でも配信ストリームとの接続に必要な範囲は改めて整理します。

一方で、ストリーミングデータそのものの基礎から学びたい方や、CloudWatch Logs・VPC Flow Logs を Firehose 経由で配信する具体的な手順を知りたい方には、本記事はやや専門的すぎる内容になります。その場合は、該当する既存記事シリーズから読み始めることをお勧めします(1-3を参照)。

チーム内での役割分担としては、配信ストリームの日々の運用・監視は想定読者であるデータ基盤エンジニア・SREが担い、Iceberg / Snowflake という新宛先の採用判断のような投資判断は、データ基盤の全体設計を統括するアーキテクトやリードエンジニアクラスの関与が望ましい進め方です。導入検討の段階から、実装を担う現場のエンジニアと、宛先の選定に責任を持つ意思決定者の両方が本記事の内容を共有しておくと、後工程での手戻りを防げます。

本記事は、Firehoseの内部実装やAWSサービス開発チームの設計意図を推測する記事ではありません。あくまで利用者の視点から、公式ドキュメントに基づく確実な設計判断を行うための実務ガイドという位置づけです。この立場は§2以降の各章、特に精度に関わる記述(改称の非対称性・リージョン対応状況・各種上限値)に一貫して反映されています。

1-3. なぜ今これを書くか

Amazon Data Firehose を主題として深掘りする記事を今このタイミングで書く理由は、大きく2つあります。

理由1: 改称とIceberg宛先追加という鮮度

2024年2月、Kinesis Data Firehose は Amazon Data Firehose へと名称変更されました。ただしこれは表示名(コンソール・ドキュメント上の名称)の変更にとどまり、API アクション名・CLI コマンド(aws firehose)・IAM アクションプレフィックス(firehose:*)・CloudWatch メトリクスの名前空間・Terraform のリソース名(aws_kinesis_firehose_delivery_stream)は、いずれも旧名(kinesis / firehose)のまま維持されています。この非対称性を正確に理解していないと、Terraform コードや IAM ポリシーを「改称に合わせて書き換えるべきもの」と誤解し、不要な移行作業を発生させてしまいます。

これに加えて、2024年10月には Apache Iceberg テーブルへの直接配信機能が追加され、同年11月には対応リージョンが拡大されました。ストリーミングデータをデータレイクの主要テーブルフォーマットである Iceberg へ直接書き込めるようになったことで、Firehose の位置づけは「S3 への単純な中継サービス」から「データレイクへの構造化された配信レイヤー」へと変わりつつあります。この変化を踏まえた本番運用の整理が、今まさに必要とされています。

理由2: 既存記事群との棲み分け — Firehose を主役化する記事の不在

本サイトには、ストリーミングデータを扱う記事がすでに複数存在しますが、いずれも Amazon Data Firehose を主題としては扱っていません。この棲み分けを整理すると次のとおりです。

いずれの既存記事も、Firehose を他サービスの宛先や配信経路として局所的に利用する場面での言及にとどまっており、Firehose 自体の配信ストリーム設計・変換ロジック・宛先選定を専門的に扱う記事は本サイトに存在しませんでした。加えて、Apache Iceberg / Snowflake という新宛先は、既存コーパス内のどの記事でも言及がなく、完全な空白領域です。本記事はこの空白を埋め、Firehose を主役に据えた初めての記事として位置づけられます。

なお、動的パーティショニングや Lambda 変換そのものの基礎的な仕組み、CloudWatch Logs・VPC Flow Logs をソースとした具体的なユースケースの詳細手順は、上記の既存記事群に委譲し、本記事では扱いません。本記事はあくまで Firehose という配信ストリーム自体の設計・宛先選定・本番運用に焦点を絞ります。

検知・分析基盤の整備が「入口(データ収集)」と「出口(可視化・分析)」の両端から進む一方で、その間をつなぐ配信ストリームそのものの設計は見落とされがちです。Firehose を主役に据えて棲み分けを整理することは、既存のストリーミング関連記事群を読み進める際の地図としても機能します。読者は本記事を通じて、自組織のデータパイプラインのどの部分に Firehose が位置し、どこから先を既存記事群に委ねればよいかを判断できるようになります。

この2つの理由(改称・新宛先追加という鮮度、Firehose を主役化する記事の不在)は、それぞれ独立した動機ではありません。改称によってFirehoseが「Kinesis Data Streamsの付属機能」から独立したサービスへとブランド上も位置づけ直され、Iceberg / Snowflakeという新宛先の追加によって実質的な機能面でも独立性が強まった、という一連の流れの中に本記事は位置しています。既存記事群がFirehoseを局所的にしか扱ってこなかったのは、これまでFirehose自体が独立した主題として扱うだけの拡張性を持っていなかった、という時系列上の理由もあります。2024年の一連のアップデートは、その状況を変える転換点でした。

AWS公式 Firehose 宛先設定リファレンス(Iceberg / Snowflakeを含む)を見る →

2. 前提・環境・準備

2-1. Amazon Data Firehose とは(改称の整理)

Amazon Data Firehoseは、ストリーミングデータをリアルタイムで収集し、変換した上でS3・Redshift・OpenSearch・Splunk・Apache Iceberg対応テーブル・Snowflakeなどの宛先へ配信するフルマネージドサービスです。サーバーやクラスターの管理は不要で、配信ストリーム(Delivery Stream)を定義するだけでスケーラブルな取り込みパイプラインを構築できます。

このサービスは2024年2月9日に、従来の「Amazon Kinesis Data Firehose」から「Amazon Data Firehose」へ正式に改称されました。「Kinesis」の名を外し、Kinesis Data Streamsとは独立したデータ配信サービスであることを明確化する狙いがあります。本記事では以降、新名称「Amazon Data Firehose」(または単に「Firehose」)を統一して使用します。

注意すべきは、表示名(ブランド名)が変わった一方で、開発者が実際に触れる技術的な識別子は一切変更されていない、という非対称な点です。

なぜAWSはこのタイミングで改称に踏み切ったのでしょうか。背景には、Amazon Data Firehoseが当初のユースケースであった「Kinesis Data Streamsのデータをバッチ配信する」役割にとどまらず、Direct PUTやAmazon MSKなど多様なソースを受け付ける独立した「データ配信サービス」へと拡張してきた経緯があります。Iceberg対応テーブルやSnowflakeといった新しい宛先の追加もこの流れの一環であり、サービス名から「Kinesis」を外すことで、Kinesis Data Streamsに依存しない独立サービスであることをブランド上も明確化したと理解しておくとよいでしょう。

読者が実務で確認すべき点を整理すると、次のとおりです。

改称にまつわる疑問として、実務でよく挙がる点をQ&A形式で整理しておきます。

Q. 旧名「Kinesis Data Firehose」で検索しても情報は見つかりますか。

見つかります。改称前に公開された公式ドキュメントのアーカイブや技術ブログ、Stack Overflowの回答などは旧名のまま残っており、内容自体は現行のAmazon Data Firehoseにもそのまま適用できます。

Q. 既存の配信ストリームのARNは改称の影響を受けますか。

受けません。ARNの形式(arn:aws:firehose:<region>:<account-id>:deliverystream/<name>)も、サービス識別子部分のfirehose表記も変更されていません。

Q. Terraformのstateファイルを改称に合わせて更新する必要はありますか。

不要です。リソースタイプ名(aws_kinesis_firehose_delivery_stream)・属性名がいずれも変更されていないため、terraform state mvのようなstate操作は一切必要ありません。

こうした背景から、本記事や関連するTerraformコード・CLIコマンド例では、文章中の呼称は「Amazon Data Firehose」で統一しつつ、実際のコマンド・リソース名・IAMポリシーは従来どおりfirehose表記のままとしています。両者が一致していないように見えても誤記ではありません。

2-2. 前提環境

本記事のハンズオンおよびコード例を実施するにあたり、以下の環境を前提とします。

• AWSアカウント: 配信ストリームの作成・宛先サービス(S3/Redshift/OpenSearch/Splunk/Iceberg対応テーブル)への書き込み権限を持つアカウント
• AWS CLI: v2系の最新版(aws --versionで2.15以降を推奨。Iceberg宛先などの新しいAPIパラメータに対応するため)
• Terraform: v1.5以降、AWS providerは5.x系以降(Iceberg宛先関連のリソース属性に対応したバージョン)
• リージョン: 本記事のコード例は東京リージョン(ap-northeast-1)を基準に記載します。Iceberg宛先などの新機能は主要リージョンから順次展開されるため、利用前に対象リージョンでの提供状況を確認してください
• IAMロール: 配信ストリーム自体が宛先へ書き込む際に引き受けるサービスロール(後述)

環境構築を進める前に、課金される軸の全体像も把握しておくと予算確保がスムーズです。具体的な単価は変動するため本記事では金額を記載しませんが、大きく次の観点で課金が発生します。

最新の単価は必ずAWS公式の料金ページで確認してください。バッファ設計やリトライ設計を通じたコスト最適化の具体的な考え方は§7で扱います。

上記に加えて、ソースにKinesis Data StreamsまたはAmazon MSKを選択する場合は、それぞれ次の前提が別途必要です。

• Kinesis Data Streamsをソースにする場合: 配信ストリームの作成前提として、同一リージョンに接続先のデータストリームが作成済みであること。配信ストリーム側からはコンシューマとしてストリームを読み取るため、kinesis:DescribeStream / kinesis:GetShardIterator / kinesis:GetRecords / kinesis:ListShardsのIAM権限を配信ストリームのサービスロールに付与する
• Amazon MSKをソースにする場合: 配信ストリームからMSKクラスターへ到達できるネットワーク経路(VPC内のプライベートエンドポイントまたはクラスターのパブリックアクセス設定)と、該当トピックへの読み取り権限を持つセキュリティグループ・IAM設定が必要

配信ストリームの作成・管理・データ投入それぞれに必要なIAM権限の例は以下のとおりです。最小権限の原則に従い、実際の運用では用途ごとにロールを分割することを推奨します。

上表のうちSplunk宛先だけは、他の宛先と認証の仕組みが異なる点に注意してください。Splunk宛先は、S3・Redshift・OpenSearchのようなIAMベースの権限ではなく、Splunk側で発行するHTTP Event Collector(HEC)のエンドポイントURLとトークンを配信ストリーム側に設定する方式で認証します。そのため、Splunk宛先を利用する場合はIAM権限の準備に加えて、Splunk側でのHECトークン発行という前提作業が別途必要になります。詳細は§4で扱います。

同様に、Snowflake宛先もIAMロールだけでなく、Snowflake側のユーザー・ロール・秘密鍵ペアの準備が前提として必要です。宛先ごとに認証方式が異なる点は、環境準備の段階でチェックリスト化しておくと後工程での手戻りを防げます。

配信ストリームが宛先へ書き込む際に引き受けるサービスロールの信頼ポリシーは、以下のとおりfirehose.amazonaws.comをプリンシパルとして許可する構成が基本形です。

{
  "Version": "2012-10-17",
  "Statement": [
 {
"Effect": "Allow",
"Principal": { "Service": "firehose.amazonaws.com" },
"Action": "sts:AssumeRole",
"Condition": {
  "StringEquals": { "sts:ExternalId": "<AWSアカウントID>" }
}
 }
  ]
}

宛先がS3バケットである場合、そのバケットが配信ストリームと異なるAWSアカウントに存在する構成(クロスアカウント配信)も選択できます。この場合、配信ストリームのサービスロールに対象バケットへのs3:PutObject等を許可するだけでなく、宛先バケット側のバケットポリシーでも配信ストリームのサービスロールARNを明示的に許可する必要があります。IAMロールの権限とリソース側のポリシーの両方が揃って初めて書き込みが成功する点は、単一アカウント構成との違いとして覚えておいてください。

Direct PUTでデータを投入する側のアプリケーションには、上記のサービスロールとは別に、次のような最小限の実行ポリシーを付与します。宛先への書き込み権限は含めず、あくまで配信ストリームへのデータ投入のみに絞る点がポイントです。

{
  "Version": "2012-10-17",
  "Statement": [
 {
"Effect": "Allow",
"Action": ["firehose:PutRecord", "firehose:PutRecordBatch"],
"Resource": "arn:aws:firehose:ap-northeast-1:<AWSアカウントID>:deliverystream/<配信ストリーム名>"
 }
  ]
}

宛先固有の権限(S3バケットへの書き込み、Icebergテーブルを管理するGlue Data CatalogへのアクセスなどのGlue関連権限)は、各宛先を扱う§4・§5でそれぞれ解説します。

なお、配信ストリームの作成数にはアカウント・リージョンごとにデフォルトのサービスクォータが設定されています。具体的な上限値はService Quotasコンソールの「Firehose」項目で随時確認できます。多数のマイクロサービスやログソースごとに配信ストリームを分割する設計を検討している場合は、事前にクォータ引き上げ申請の要否を確認しておくとよいでしょう。

環境構築後は、以下のようなCLIコマンドで疎通確認しておくと、本記事の後続手順をスムーズに進められます。

# 認証情報とリージョンの確認
aws sts get-caller-identity

# 対象リージョンに存在する配信ストリームの一覧を取得(権限確認を兼ねる)
aws firehose list-delivery-streams --region ap-northeast-1 --limit 10

list-delivery-streamsがエラーなく応答すれば、CLIの認証情報とfirehose:ListDeliveryStreams権限が正しく設定されていることが確認できます。Python(Boto3)から操作する場合も、boto3.client("firehose").list_delivery_streams()で同様に疎通確認が可能です。

各宛先(S3・Redshift・OpenSearch・Splunk・Iceberg対応テーブル・Snowflake)は、機能追加のタイミングによって対応リージョンにばらつきがあります。特にIceberg対応テーブルやSnowflake宛先のような比較的新しい機能は、GA後もリージョン展開が段階的に進むことがあるため、構築前に必ずAWSリージョン別サービス一覧やAWS公式ドキュメントの該当ページで、対象リージョンでの提供状況を確認する運用を前提としてください。この確認を怠ると、Terraform apply時やコンソールでの作成時にリソース作成エラーとなり、原因の切り分けに時間を要することがあります。

2-3. 配信ストリームの基本構造

Amazon Data Firehoseの中心概念は「配信ストリーム(Delivery Stream)」です。配信ストリームは、データがどこから入り(ソース)、どのようにバッファリング・変換され、どこへ届くか(宛先)を1つの単位として定義するリソースです。fig02のとおり、全体は「ソース → バッファリング → 変換 → 宛先」という一方向のパイプラインとして構成されます。

配信ストリームが受け付けるソースは、次の3種類から選択します。

いずれのソースも、配信ストリーム作成後に別のソース種別へは変更できません。ソースの選択は配信ストリームのライフサイクルを通じて固定される設計判断であるため、後から構成変更が必要になった場合は、新しい配信ストリームを作成し直す必要がある点に注意してください。

どのソースを選ぶべきかは、データの発生元と既存のストリーミング基盤の有無で決まります。新規にアプリケーションからログやイベントを送信する場合はDirect PUTが最もシンプルで、追加のインフラを必要としません。すでにKinesis Data Streamsでリアルタイム処理を行っており、その結果を追加でS3やIcebergテーブルにも永続化したい場合はKinesis Data Streamsをソースに選びます。オンプレミスや他クラウドから移行してきたKafkaベースのイベント基盤を持つ場合は、Amazon MSKをソースに指定することで、既存のプロデューサー実装を変更せずにAmazon Data Firehoseへ接続できます。

fig02は、この全体構成を4つのブロックで表しています。左端の「ソース」ブロックでは、Direct PUT・Kinesis Data Streams・Amazon MSKのいずれかからレコードが配信ストリームに投入されます。次の「バッファリング」ブロックでは、投入されたレコードが指定したサイズまたは時間のいずれか早い方の条件に達するまで一時的に蓄積されます。3つ目の「変換」ブロックは任意の処理で、Lambdaによるレコード単位の加工や、JSON形式からParquet/ORCへのフォーマット変換、Icebergテーブルへの書き込みに必要なスキーマ・パーティション振り分けの処理などがここに該当します(変換を行わずソースの形式のまま配信できます)。右端の「宛先」ブロックで、変換後のデータが指定した1つの宛先へバッチとして配信されます。

配信ストリームの命名にあたっては、後から複数の配信ストリームを見分けやすいよう、用途・環境・宛先が分かる名前を付けることを推奨します。例えばapp-logs-prod-to-icebergのように「データ種別 – 環境 – 宛先」の3要素を含めておくと、CloudWatchメトリクスやコスト集計を宛先別に追いやすくなります。配信ストリーム名は作成後の変更ができないため、命名規則はチーム内で事前に取り決めておくとよいでしょう。

配信ストリームの基本構造を理解するうえで、押さえておくべき制約が2つあります。1つ目は、1つの配信ストリームは1つのソースと1つの宛先のみを持つ、シンプルな1:1構造だという点です。同じデータを複数の宛先に同時配信したい場合は、単一の配信ストリームでは実現できません。Kinesis Data Streamsをソースにした上で複数の配信ストリームをファンアウトさせる、あるいは宛先ごとに個別のソース経路を用意するといった設計が必要になります。

2つ目は、S3以外の宛先(Redshift・OpenSearch・Splunk)を選択した場合でも、内部的にはS3が中継地点として使われるという点です。これらの宛先では、配信に失敗したレコードやバックアップ用のコピーがS3バケットへ書き込まれる仕組みです。宛先設定時には、S3バケットの指定も併せて必要になります。Icebergテーブル宛先の場合も、テーブルの実体はS3上に構築されるため、S3バケットとの関係は切り離せません。この点の具体的な設定は§4・§5で扱います。

投入するデータの形式についても触れておきます。Direct PUTでは任意のバイナリペイロード(レコード1件あたり最大1,000KB)を投入できますが、実運用ではJSON形式(改行区切りのNDJSON)を使うケースが大半です。変換ブロックでのフォーマット変換(Parquet/ORC化)やIcebergテーブルへのマッピングは、投入データがJSON形式であることを前提とした機能が中心となるため、ソース側でどのようなデータ形式を送出するかは、後段の変換・宛先設計にも影響します。

配信ストリームを作成すると、内部的には次のような状態遷移をたどります。作成直後はCREATING状態となり、宛先やIAMロールの設定が正常に反映されるとACTIVE状態に遷移してデータ投入が可能になります。設定に誤りがある場合はCREATING_FAILED状態となり、DescribeDeliveryStream APIのレスポンスに含まれるエラーメッセージから原因を特定できます。削除時も同様にDELETINGを経て完全に削除される流れです。

Terraformで管理している場合も、terraform apply完了後にコンソールまたはaws firehose describe-delivery-stream --delivery-stream-name <名前>でDeliveryStreamStatusがACTIVEになっていることを確認してから、データ投入や後続の宛先設定検証に進むことを推奨します。

いずれのソースを選んだ場合も、配信ストリームに投入されたデータはバッファ(バッファサイズまたはバッファインターバルのいずれか早く条件を満たした時点でフラッシュ)に蓄積され、必要に応じてLambdaによる変換やフォーマット変換(JSON→Parquet/ORCなど)を経て、宛先へ配信されます。バッファの具体的なサイズ・インターバル設計やエラーハンドリング、監視の詳細は§7で、Lambdaによるデータ変換とフォーマット変換の詳細は§6で扱います。

なお、CloudWatch LogsのサブスクリプションフィルタやVPC Flow Logsのように、特定のAWSサービスからAmazon Data Firehoseへログを転送するソース別の具体的なユースケースは、本記事のスコープ外とします。これらは既に「AWS Observability本番運用 Vol2(CloudWatch Logs深掘り編)」や「AWSネットワーク可観測性 Vol1」で解説済みのため、該当する設定・運用ノウハウはそちらに委譲し、本記事では配信ストリーム自体の設計・変換・宛先(特にIceberg/Snowflakeという新宛先)に焦点を絞ります。

配信ストリームが対応する宛先は、S3・Redshift・OpenSearch(Service/Serverless)・Splunk・HTTPエンドポイント、そして新しく追加されたApache Iceberg対応テーブル・Snowflakeまで多岐にわたります。これら宛先ごとの設定と選定基準は§4で、本記事の差別化の核となるIceberg/Snowflake宛先の詳細は§5で扱います。ここでは、どのソースを選んでも同じ配信ストリームの枠組み(バッファ→変換→宛先)に乗る、という全体像を押さえておいてください。

ここまでで、改称の経緯と技術的な非対称性、環境構築時に必要なIAM権限・クォータ・課金軸、そして配信ストリームの基本構造(ソース→バッファリング→変換→宛先の1:1パイプライン)を整理しました。次の§3では、この基本構造のうち特に「ソース」部分に焦点を当て、Direct PUT・Kinesis Data Streams・Amazon MSKそれぞれの具体的な設定手順とIAMロール構成を実装レベルで解説します。

3. 配信ストリームとソース設定

Amazon Data Firehose の配信ストリームは、作成時にソースを Direct PUT・Amazon Kinesis Data Streams・Amazon MSK の3種類から1つ選択します。ソース種別によって認証方式・スループットの上限・IAM ロールの構成が異なるため、それぞれの設定要点を実装レベルで整理します。

3-1. Direct PUT — アプリケーションから直接書き込む

Direct PUT は、producer アプリケーションが PutRecord / PutRecordBatch API を直接呼び出して配信ストリームにデータを書き込む方式です。AWS SDK・AWS Lambda・CloudWatch Logs / Events・Amazon SNS・AWS WAF web ACL ログ・Amazon API Gateway アクセスログ・Kinesis Agent など、多数の AWS サービス/エージェントが Direct PUT に対応しています。

producer に付与する IAM ポリシーは、対象の配信ストリーム ARN に絞り込みます。

{
 "Version": "2012-10-17",
 "Statement": [
  {
"Effect": "Allow",
"Action": [
 "firehose:PutRecord",
 "firehose:PutRecordBatch"
],
"Resource": [
 "arn:aws:firehose:ap-northeast-1:123456789012:deliverystream/delivery-stream-name"
]
  }
 ]
}

Direct PUT には API レベルのクォータがあります。設計時に必ず確認してください。

スループットが上限に達すると Firehose は自動的にストリームのスループット上限を引き上げますが、引き上げが完了するまではスロットリングが発生するため、producer 側でリトライ処理を実装しておく必要があります。急激なバースト増加が見込まれる場合は、事前に Service Quotas からスループット引き上げをリクエストしてください。

3-2. Kinesis Data Streams をソースにする

既存の Kinesis Data Stream を配信ストリームのソースとして選択すると、Firehose がそのストリームからレコードを自動的に読み取り、宛先へロードします。producer 側は通常どおり Kinesis Data Streams に書き込むだけでよく、Firehose を意識した実装変更は不要です。

Firehose の配信ロールには、ソースストリームに対する読み取り権限が必要です。

{
 "Effect": "Allow",
 "Action": [
  "kinesis:DescribeStream",
  "kinesis:GetShardIterator",
  "kinesis:GetRecords",
  "kinesis:ListShards"
 ],
 "Resource": "arn:aws:kinesis:ap-northeast-1:123456789012:stream/stream-name"
}

Kinesis Data Streams をソースにした場合、Direct PUT のスループットクォータは適用されず、Firehose は上限なくスケールします。また宛先が利用不可になった場合のデータ保持は、Firehose の24時間固定ではなくソース側の Kinesis Data Streams の保持設定に従います。

シャード数の設計・KPL による集約・パーティションキー設計など、Kinesis Data Streams 自体の運用ノウハウは既存の Kinesis Data Streams 本番運用記事で解説しています。本記事は Firehose 側のソース設定に絞って扱います。

3-3. Amazon MSK をソースにする

Amazon MSK(プロビジョンド / Serverless)をソースにすると、指定した Kafka トピックから Firehose がレコードを読み取り、S3 系の宛先へロードします。トピック指定は配信ストリーム作成後に変更できないため、事前に確定させてください。

クラスター接続方式

• Private bootstrap brokers(推奨): クラスターが Active であること、アクセス制御に IAM が含まれること、Multi-VPC private connectivity が有効であること、Firehose サービスプリンシパルに CreateVpcConnection API 呼び出しを許可するリソースベースポリシーがクラスターに付与されていることが条件です。
• Public bootstrap brokers: プロビジョンドクラスターのみ対応で、Active・IAM アクセス制御あり・パブリックアクセス可能である必要があります。

IAM ロールに必要な権限

{
"Effect": "Allow",
"Action": [
"kafka:GetBootstrapBrokers",
"kafka:DescribeCluster",
"kafka:DescribeClusterV2",
"kafka-cluster:Connect"
],
"Resource": "arn:aws:kafka:ap-northeast-1:123456789012:cluster/cluster-name/cluster-uuid"
},
{
"Effect": "Allow",
"Action": [
"kafka-cluster:DescribeTopic",
"kafka-cluster:DescribeTopicDynamicConfiguration",
"kafka-cluster:ReadData"
],
"Resource": "arn:aws:kafka:ap-northeast-1:123456789012:topic/cluster-name/cluster-uuid/topic-name"
},
{
"Effect": "Allow",
"Action": ["kafka-cluster:DescribeGroup"],
"Resource": "arn:aws:kafka:ap-northeast-1:123456789012:group/cluster-name/cluster-uuid/*"
}

MSK ソースには専用のクォータがあります。パーティションあたりの読み取りスループットは10 MB/秒、レコードサイズは最大10 MBです。ただし Lambda によるデータ変換を有効にしている場合、Lambda 自体の受信レコードサイズ上限が6 MBのため、6 MBを超えるレコードはエラー用 S3 バケットに配信されます。大きめのレコードを扱う設計では、Lambda 変換の要否とレコードサイズ上限の関係を必ず確認してください。

Firehose に共通のロール引き受け設定

ソース種別によらず、配信ストリームが使用する IAM ロールには次の信頼ポリシーが必要です。

{
  "Version": "2012-10-17",
  "Statement": [
 {
"Principal": { "Service": ["firehose.amazonaws.com"] },
"Effect": "Allow",
"Action": "sts:AssumeRole"
 }
  ]
}

MSK をソースにする場合は、宛先アクセス用のロールとは別に、MSK クラスターからのデータ取り込みを許可する専用ロールを用意し、同じ信頼ポリシーを設定します。

4. 宛先設定 — S3 / Redshift / OpenSearch Service / Splunk

宛先ごとに設定項目・IAM 要件・エラー時の挙動が異なります。本節では S3・Redshift・OpenSearch Service・Splunk の4宛先について、実装レベルの設定と実務上の注意点を扱います。

4-1. Amazon S3 — プレフィックス設計と動的パーティショニングの実務

S3 宛先では、バケット・S3プレフィックス・圧縮方式・暗号化に加えて、動的パーティショニングの設計が本番運用の要になります。

Firehose はレコードを次の順序で処理します。

1. KPL(protobuf)による deaggregation
2. JSON またはデリミタによる multi record deaggregation
3. Lambda によるデータ変換
4. 動的パーティショニング(データのパーティション分け)
5. データフォーマット変換(Parquet/ORC 等)
6. S3 への配信

複数パーティションキーの実務

動的パーティショニングでは、inline parsing(jqクエリ)または Lambda 関数のいずれか、あるいは両方を組み合わせてパーティショニングキーを生成できます。1配信ストリームあたり最大50個のパーティショニングキーを指定可能です。S3 バケットプレフィックスは、inline parsing 由来のキーなら !{partitionKeyFromQuery:keyID}、Lambda 由来のキーなら !{partitionKeyFromLambda:keyID} の形式で記述し、キー数とプレフィックス数を一致させる必要があります。

例えば、レコードが {"customer_id": "c001", "event_type": "purchase", "region": "apne1"} のようなJSONで、customer_id と event_type の2キーでパーティショニングする場合、inline parsingのキー/値には次のようなjq式(Firehoseはjq 1.6のみサポート)を指定します。

対応するS3バケットプレフィックスは data/!{partitionKeyFromQuery:customerId}/!{partitionKeyFromQuery:eventType}/ のように記述します。集約済みレコード(KPLでの集約やPutRecordBatchでの複数イベント同梱)を扱う場合は、multi record deaggregationを有効にしたうえで動的パーティショニングを設定しないと、集約単位でしかパーティション分けできない点に注意してください。

動的パーティショニングのクォータとエラー処理

動的パーティショニングを有効にした配信ストリームには、デフォルトで最大500個のアクティブパーティション、パーティションあたり最大1 GB/秒のスループットというクォータがあります。バッファ間隔とパーティション生成レートから想定アクティブパーティション数を事前に見積もり、上限に近い場合は配信ストリームを分割してパーティションを分散させます。

jq 式が対象レコードに一致しない、または不正な JSON を受け取った場合、そのレコードはパーティショニングに失敗し、設定した S3 error output prefix に配信されます。本番運用では、error prefix を監視対象に含め、失敗レコードを定期的に再処理(re-drive)する運用を組み込んでください。

動的パーティショニングの基本的なコード例(jq式の書き方・Lambda関数の実装例)は、データ分析基盤の既存記事で解説しています。本記事では複数パーティションキー設計とエラー処理という発展的な内容に絞って扱います。

IAM ポリシー(S3宛先)

{
 "Effect": "Allow",
 "Action": [
  "s3:AbortMultipartUpload",
  "s3:GetBucketLocation",
  "s3:GetObject",
  "s3:ListBucket",
  "s3:ListBucketMultipartUploads",
  "s3:PutObject"
 ],
 "Resource": [
  "arn:aws:s3:::amzn-s3-demo-bucket",
  "arn:aws:s3:::amzn-s3-demo-bucket/*"
 ]
}

S3バケットの所有者がFirehose実行アカウントと異なる場合は s3:PutObjectAcl を追加し、バケット所有者にオブジェクトのフルアクセスを付与します。

4-2. Amazon Redshift — 中間S3経由のCOPY連携

Redshift 宛先では、Firehose はまずデータを中間 S3 バケットに配信し、その後 Redshift の COPY コマンドを発行してテーブルへロードします。プロビジョンドクラスターと Redshift Serverless ワークグループのどちらも宛先にできますが、パブリックアクセス可能であることが共通の前提条件です(Firehose の IP アドレスを許可リストに追加する必要があります)。

主な設定項目は次のとおりです。

• 認証: ユーザー名/パスワードを直接指定するか、AWS Secrets Manager のシークレットを参照します。指定するRedshiftユーザーには、S3からテーブルへのデータロードに必要な INSERT 権限が必要です。
• COPY options / COPY command: GZIP(S3圧縮を有効にしている場合は必須)や REGION(S3バケットとRedshiftクラスターのリージョンが異なる場合に必須)などのパラメータを指定します。
• Retry duration: COPY 失敗時のリトライ時間(0〜7,200秒)。5分間隔でリトライし、期間終了後は配信失敗として扱われます。

IAM ロールには S3 宛先と同様のバケットアクセス権限に加えて、KMS暗号化・CloudWatch Logsへのエラー出力・Lambda変換用の権限を付与します。Redshift 自体へのアクセスはIAMではなく、指定したユーザー名/パスワードまたはSecrets Managerのシークレットで行われる点に注意してください。

4-3. OpenSearch Service — インデックスローテーションとドキュメントID

OpenSearch Service 宛先では、ドメイン・インデックス名に加えて次の項目を設定します。

• Index rotation: インデックスをローテーションする場合、Firehose がインデックス名にタイムスタンプを付与します。ログ量に応じてローテーション間隔を選択してください。
• DocumentID type: Firehose生成のドキュメントIDとOpenSearch Service生成のドキュメントIDを選択できます。書き込み負荷の高いログ分析・オブザーバビリティ用途では、OpenSearch Service側でCPU消費を抑えられるOpenSearch Service生成方式が推奨されます。
• Retry duration: インデックスリクエスト失敗時のリトライ時間で、デフォルトは300秒、最大7,200秒まで設定可能です。リトライ期間が終了すると、データはDead Letter Queue(設定したS3エラーバケット)に配信され、そこから手動でOpenSearch側へ再投入(re-drive)する必要があります。

ドメインがVPC内にある場合、Firehoseの配信ロールに ec2:DescribeVpcs / ec2:DescribeSubnets / ec2:DescribeSecurityGroups / ec2:CreateNetworkInterface / ec2:CreateNetworkInterfacePermission / ec2:DeleteNetworkInterface などENI操作権限を追加で付与する必要があります。これらの権限は配信ストリーム作成後も取り消さないでください。取り消すとENIの作成・更新ができなくなり、配信が停止します。

4-4. Splunk — HECエンドポイントとRaw/Event設定

Splunk 宛先では、Firehose は Splunk の HTTP Event Collector(HEC)エンドポイントにデータを送信します。ロードバランサーはClassic Load BalancerまたはApplication Load Balancerのみサポートされ、Duration-Basedのスティッキーセッションを有効にする必要があります。

• Splunk endpoint type: 通常は Raw endpoint を選択します。Lambdaで前処理し、イベント種別ごとに異なるインデックスへ振り分ける場合は Event endpoint を選択します。
• Authentication: HECトークンを直接入力するか、Secrets Managerのシークレットを参照します。
• HEC acknowledgement timeout: SplunkからのインデックスACKをFirehoseが待機する時間です。タイムアウトすると配信失敗として扱われます。
• Retry duration: ACKタイムアウトまたはエラー発生後にFirehoseがリトライする時間です。リトライのたびにACKタイムアウトのカウンターがリセットされる点に注意してください。リトライ期間が終了すると、データはS3バックアップバケットに保存されます。

Firehose はSplunkへのアクセスにIAMを使用せず、HECトークンで認証します。そのためIAMロールに付与するのは、S3バックアップバケットへのアクセス・CloudWatch Logsのエラー出力・Lambda変換用の権限のみで、Splunkリソース自体に対するIAM権限は不要です。

4-5. 宛先ごとのリトライ・バックアップ挙動の比較

4宛先とも「配信失敗時は最終的にS3へバックアップする」という共通の設計思想を持ちますが、リトライ時間の可変範囲と失敗データの戻し先には違いがあります。設計時は次の一覧で挙動の差分を押さえておいてください。

いずれの宛先でも、S3に退避されたデータは自動では宛先へ再投入されません。監視(CloudWatchメトリクス・S3イベント通知など)と再処理(re-drive)の運用フローは§7で扱います。

5. Apache Iceberg / Snowflake 宛先(新宛先)

§4では S3・Redshift・OpenSearch・Splunk という従来からの宛先を扱いました。本節では、2024年に追加された Apache Iceberg テーブル と Snowflake という2つの新宛先を扱います。どちらも「配信するだけ」ではなく、テーブル形式のデータストアへ直接 insert/update/delete を反映できる点が従来宛先との決定的な違いです。

5-1. Apache Iceberg tables 宛先

Apache Iceberg は、S3データレイクにSQLテーブルの信頼性とシンプルさをもたらすオープンソースのテーブルフォーマットです。Spark・Flink・Trino・Hive・Impalaなど複数の分析エンジンから同一データへ同時アクセスできる点が特徴です。Firehoseは2024年10月にIceberg宛先をGAし、2024年11月には対応リージョンを拡大しました。

Iceberg宛先の設定は IcebergDestinationConfiguration で行います。主要パラメータは以下の通りです。

DestinationTableConfigurationList の各要素(DestinationTableConfiguration)では、DestinationDatabaseName・DestinationTableName に加えて、update/delete操作するテーブルには UniqueKeys の指定が必須です。

{
  "IcebergDestinationConfiguration": {
 "RoleARN": "arn:aws:iam::123456789012:role/FirehoseIcebergRole",
 "CatalogConfiguration": {
"CatalogARN": "arn:aws:glue:ap-northeast-1:123456789012:catalog"
 },
 "S3Configuration": {
"RoleARN": "arn:aws:iam::123456789012:role/FirehoseIcebergRole",
"BucketARN": "arn:aws:s3:::my-iceberg-datalake"
 },
 "DestinationTableConfigurationList": [
{
  "DestinationDatabaseName": "iot_events",
  "DestinationTableName": "device_readings",
  "UniqueKeys": ["deviceId", "timestamp"]
}
 ],
 "AppendOnly": false,
 "BufferingHints": { "SizeInMBs": 128, "IntervalInSeconds": 300 }
  }
}

データベース変更の継続レプリケーション(2024-11・プレビュー)

2024年11月には、MySQL/PostgreSQLなど運用データベースの変更をFirehose経由で継続的にApache Icebergテーブルへレプリケーションする機能がプレビューで追加されました。データベースのレプリケーションログを直接利用してCDC(Change Data Capture)ストリームを取得するため、データベース本体のトランザクション性能への影響を抑えられる設計です。Firehoseは対象テーブルの初期フルコピーを実施した後、継続的にinsert/update/deleteの変更をIcebergテーブルへ反映します。本記事執筆時点ではプレビュー機能のため、本番導入前に最新のサービス状況を公式ドキュメントで確認してください。

5-2. Snowflake 宛先

Snowflakeは、Firehoseコンソールから接続情報を設定するだけで組み込み宛先として利用できます。Snowpipe StreamingのAPIを自前で呼び出すコードを書く必要はなく、Firehoseの認証情報を設定するだけでストリーミングロードを完結できる点が特徴です。

接続設定

秘密鍵はPEMヘッダー・フッターと改行を除去した状態で指定する必要があります。

-----BEGIN PRIVATE KEY-----
KEY_CONTENT
-----END PRIVATE KEY-----

上記から KEY_CONTENT の部分(改行を除去したもの)のみをFirehoseの秘密鍵フィールドに設定します。鍵ローテーションを運用する場合はSecrets Manager経由での参照に統一しておくと切り替えが容易です。

データベース設定とデータロードオプション

Snowflake宛先では、宛先データベース・スキーマ・テーブルを指定した上で、2種類のデータロード方式から選択します。

VARIANTカラムでMetadata columnを有効にすると、ソース種別に応じて以下のようなメタデータが自動付与されます。

// Direct PUT がソースの場合
{
  "firehoseDeliveryStreamName": "my-delivery-stream",
  "IngestionTime": "2026-07-09T10:00:00Z"
}

// Kinesis Data Streams がソースの場合
{
  "kinesisStreamName": "my-source-stream",
  "kinesisShardId": "shardId-000000000001",
  "kinesisPartitionKey": "device-1234",
  "kinesisSequenceNumber": "49655962066601463032522589543535113056108699331451682818000000",
  "subsequenceNumber": "1",
  "IngestionTime": "2026-07-09T10:00:00Z"
}

Retry durationとBuffering hintsは他の宛先と同様の考え方で設定します。Snowflake接続やチャネルオープンに失敗した場合、Retry durationで指定した時間だけ指数バックオフでリトライし、それでも失敗した場合はS3エラーバケットへデータを退避します。

5-3. 新宛先の実践例 — JSON式ルーティングの設定コード

Iceberg宛先の最大の特徴は、単一の配信ストリームから複数のテーブルへレコードを振り分けられる点です。振り分け方法は2通り用意されています。

方法1: JSONQuery式による振り分け(推奨・低コスト)

Database Name・Table Name・Operationの3パラメータに、それぞれ静的な値または受信レコードから値を取り出す動的な式を指定します。例えば、以下の受信レコードの deviceId フィールドの値によって書き込み先テーブルを振り分ける場合です。

{
  "deviceId": "Device1234",
  "timestamp": "2024-11-28T11:30:00Z",
  "data": { "temperature": 21.5 },
  "status": "online"
}

Database Name : "iot_events"
Table Name : .deviceId
Operation  : "insert"

deviceId がネストしている場合は .event.deviceId のようにパスで指定します。この方式は追加のLambda実行コストがかからず、単純な振り分けルールであれば最もコスト効率が良い選択肢です。

方法2: Lambda変換関数による振り分け(複雑なルール向け)

フィールド値による条件分岐や、レコードへの属性追加を伴う複雑な振り分けが必要な場合は、Lambda変換関数の出力に metadata.otfMetadata セクションを追加して振り分け先を指定します。

firehose_record_output = {
 "recordId": firehose_record_input["recordId"],
 "result": "Ok",
 "data": base64.b64encode(payload_bytes.encode("utf-8")),
 "metadata": {
  "otfMetadata": {
"destinationDatabaseName": "iot_events",
"destinationTableName": "device_readings",
"operation": "insert",
  }
 },
}

destinationDatabaseName と destinationTableName が必須で、operation は省略可能です(未指定時のデフォルトは insert)。JSONQuery式とLambda変換の両方を設定している場合、Firehoseはまず Lambda出力の metadata を優先して参照し、不足しているフィールドのみJSONQuery式の結果で補完します。

5-4. 東京リージョンでの利用に関する注意点

新宛先の採用可否は対応リージョンの確定が前提になるため、公式ドキュメントで最終確認した結果を記載します。

Apache Iceberg tables 宛先: Firehose公式ドキュメントによれば、Firehoseは中国リージョン・AWS GovCloud (US)・アジアパシフィック(台北)・アジアパシフィック(マレーシア)・アジアパシフィック(ニュージーランド)・メキシコ(中部)を除く全リージョンでIceberg宛先をサポートしています。東京リージョン(ap-northeast-1)はこの除外リストに含まれないため利用可能です。

Snowflake 宛先: 同ドキュメントでは、Snowflake宛先の対応リージョンが個別に列挙されており、アジアパシフィック(東京)は明示的にサポート対象リージョンとして記載されています(米国東部(バージニア北部)・米国西部(オレゴン)・欧州(アイルランド)・米国東部(オハイオ)・欧州(フランクフルト)・アジアパシフィック(シンガポール/ソウル/シドニー/ムンバイ/大阪/ジャカルタ)・欧州(ロンドン/パリ/ストックホルム)・南米(サンパウロ)・カナダ(中部)など)。

6. Lambda データ変換と Format Conversion

6-1. Lambda データ変換の設定 — バッファリングとペイロード上限

配信ストリームには変換用の AWS Lambda 関数を組み込めます。有効化すると、Firehose は取り込んだレコードをバッファリングしたうえで Lambda を同期呼び出し(Synchronous Invocation)し、変換結果を受け取ってから宛先バッファへ送ります。この変換用バッファリングは、宛先へ配信する際のバッファリング(7章で扱います)とは別の設定値を持ちます。

Lambda の同期呼び出しには リクエスト・レスポンス合計で 6 MB のペイロード上限 があります。バッファサイズを大きく取りすぎると、Lambda への送信ペイロードや変換後レスポンスがこの上限を超え、超過分のレコードはエラー出力(7章参照)へ振り分けられて配信失敗します。1 レコードが肥大化しやすいユースケースでは BufferSizeInMBs を保守的に(1–2 MB 程度)設定するのが安全です。

{
  "ProcessingConfiguration": {
 "Enabled": true,
 "Processors": [
{
  "Type": "Lambda",
  "Parameters": [
 { "ParameterName": "LambdaArn", "ParameterValue": "arn:aws:lambda:ap-northeast-1:123456789012:function:firehose-transform" },
 { "ParameterName": "BufferSizeInMBs", "ParameterValue": "2" },
 { "ParameterName": "BufferIntervalInSeconds", "ParameterValue": "60" },
 { "ParameterName": "NumberOfRetries", "ParameterValue": "3" }
  ]
}
 ]
  }
}

NumberOfRetries はネットワークタイムアウトや Lambda 同時実行数上限到達時のリトライ回数で、デフォルトは 3 回です。3 回とも失敗すると、そのバッチは処理失敗レコードとして扱われます(詳細は 7 章)。

6-2. Lambda 関数側の実装 — ステータスコードとレコード形式

Firehose は Lambda のレスポンスに含まれる各レコードの result フィールドで処理結果を判定します。

recordId は入力レコードと出力レコードで必ず一致させる必要があります。省略や不一致があると Firehose 側でエラーになります。

import base64
import json

def lambda_handler(event, context):
 output = []
 for record in event["records"]:
  payload = json.loads(base64.b64decode(record["data"]))
  try:
payload["ingested_at"] = record["approximateArrivalTimestamp"]
result = "Ok"
data = base64.b64encode(json.dumps(payload).encode("utf-8")).decode("utf-8")
  except Exception:
result = "ProcessingFailed"
data = record["data"]
  output.append({"recordId": record["recordId"], "result": result, "data": data})
 return {"records": output}

6-3. Format Conversion — Glue Schema による JSON→Parquet/ORC 変換

Format Conversion を使うと、S3 へ格納する直前に JSON を Apache Parquet または Apache ORC へ変換できます。設定には以下 3 要素が必須です。

1. デシリアライザ: 入力 JSON の解析方式。タイムスタンプ形式が多様な場合は OpenX JSON SerDe、uniontype など特殊な型を扱う場合は Hive JSON SerDe を選択します。
2. スキーマ: AWS Glue Data Catalog に事前登録したテーブル定義を参照します。スキーマと実データの構造が一致しない場合、スキーマにない属性は変換後データに含まれません。
3. シリアライザ: 出力先フォーマットに応じて ORC SerDe または Parquet SerDe を選択します。

サイズ上限にも注意が必要です。長さ指定のない型を含む 1 行あたりのデータ量には、実用上 32 MB という上限があります。超過するとその行は変換に失敗します。

{
  "DataFormatConversionConfiguration": {
 "Enabled": true,
 "InputFormatConfiguration": {
"Deserializer": { "OpenXJsonSerDe": {} }
 },
 "OutputFormatConfiguration": {
"Serializer": { "ParquetSerDe": {} }
 },
 "SchemaConfiguration": {
"DatabaseName": "prod_datalake",
"TableName": "clickstream_raw",
"RoleARN": "arn:aws:iam::123456789012:role/firehose-glue-role"
 }
  }
}

6-4. 動的パーティショニングの発展的活用 — 複数パーティションキーと JQ 式

動的パーティショニングの基礎(有効化手順・単一キーでの S3 プレフィックス自動分割)は、Data Analytics 本番運用 Vol2 の Kinesis Data Streams 本番運用 §2-5 で解説済みのため、本節では複数パーティションキーの組み合わせとエラー処理という発展的な内容を扱います。

インライン解析(jq 1.6)では、ネストされたフィールドや日時のフォーマット変換を組み合わせて複数キーを同時に定義できます。

S3 プレフィックス例:

Prefix: customer_id=!{partitionKeyFromQuery:customer_id}/event_type=!{partitionKeyFromQuery:event_type}/year=!{partitionKeyFromQuery:year}/month=!{partitionKeyFromQuery:month}/day=!{partitionKeyFromQuery:day}/
ErrorOutputPrefix: errors/!{firehose:error-output-type}/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/

キー数を増やすほどアクティブパーティション数が増加する点に注意してください。アクティブパーティション数にはデフォルトで 500 という上限があり、超過が近づくと PartitionCountExceeded メトリクスが 1 を返します。上限に達しそうな場合は、配信ストリームを分割してパーティションを分散させるか、バッファリング間隔を短くしてアクティブ状態のパーティションを早期に確定させます。

jq 式の評価に失敗したレコードやパーティションキーの値が取得できなかったレコードは、宛先への配信対象から外れ、ErrorOutputPrefix で指定した S3 エラー出力先へエラー種別ごとに退避されます。動的パーティショニングを有効化する場合、ErrorOutputPrefix の指定は必須です。

6-5. Parquet と ORC の選定基準

Format Conversion で JSON をどちらの列指向フォーマットへ変換するかは、後段の分析基盤との組み合わせで判断します。

新規にデータレイクを構築するケースでは、Athena・EMR Serverless・QuickSight など AWS 分析サービス全般との親和性から Parquet を既定選択とし、既存基盤が ORC 前提の場合のみ ORC を選ぶ、という判断が実務上シンプルです。

6-6. 変換パイプラインの実務ポイント — べき等性とオブザーバビリティ

Lambda 変換関数は、Firehose のリトライによって 同一レコードが複数回実行される ことも珍しくありません。関数内で外部システムへの副作用(DB書き込み・外部API呼び出しなど)を伴う場合は、recordId を用いた重複排除やべき等な処理設計が必須です。

また、変換ロジックの不具合は Format Conversion のスキーマ不整合と組み合わさると原因切り分けが難しくなります。Lambda 関数内で構造化ログ(レコードの一部・errorCode 相当の分類ラベル)を CloudWatch Logs に出力しておくと、7 章で扱う FailedConversion.Records の増加時にログを起点とした原因調査が可能になります。

7. バッファリング・エラーハンドリング・監視・コスト最適化

7-1. バッファリング設計 — サイズ上限 128 MB・インターバル上限 900 秒

宛先(S3 等)へ配信する際のバッファリングヒントは、6 章で扱った Lambda 変換用バッファリングとは別の値です。

Firehose はサイズ・間隔のどちらか早く条件を満たした時点で配信します。リアルタイム性を重視する場合はサイズを小さく・間隔を短く(例: 5 MB / 60 秒)、コストと S3 オブジェクト数を抑えたい場合はサイズを大きく・間隔を長く(例: 128 MB / 900 秒)設定します。特に Format Conversion や動的パーティショニングを併用する場合、バッファ間隔を短くしすぎると 1 パーティションあたりの配信オブジェクト数が増え、小さな Parquet/ORC ファイルが大量発生してAthena 等のクエリ性能を悪化させる点に注意してください。

7-2. S3 バックアップと ErrorOutputPrefix によるエラー処理

配信に失敗したレコードは ErrorOutputPrefix で指定した S3 プレフィックスへ、エラー種別(error-output-type)ごとに退避されます。エラー種別には processing-failed(Lambda 変換失敗)・format-conversion-failed(Parquet/ORC 変換失敗)・宛先サービスごとの *-failed などがあり、ErrorOutputPrefix には必ず !{firehose:error-output-type} を含める必要があります。

Prefix: data/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/
ErrorOutputPrefix: errors/!{firehose:error-output-type}/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/

Lambda 変換が失敗した場合、processing-failed フォルダには以下の形式でレコードが格納されます。rawData に元データが Base64 で残るため、再処理パイプラインの入力としてそのまま利用できます。

{
  "attemptsMade": "3",
  "arrivalTimestamp": "1720500000000",
  "errorCode": "Lambda.FunctionError",
  "errorMessage": "...",
  "attemptEndingTimestamp": "1720500005000",
  "rawData": "eyJjdXN0b21lcl9pZCI6ICIxMjM0In0=",
  "lambdaArn": "arn:aws:lambda:ap-northeast-1:123456789012:function:firehose-transform"
}

Lambda 呼び出し自体がネットワークタイムアウトや同時実行数上限により失敗した場合、Firehose はデフォルトで 3 回リトライし、それでも失敗すればそのバッチを処理失敗レコードとして扱います。リトライ回数は 6 章で示した NumberOfRetries パラメータを使うと調整できます。

S3 バックアップには 2 つのモードがあります。データ変換を有効化している場合に選択でき、失敗レコードのみを退避する FailedDataOnly(既定)と、変換前の全レコードを別バケットへ複製する AllData があります。監査要件や、変換ロジックの不具合発生時に元データから再処理したいケースでは AllData を検討してください。ただし全レコードを複製するため、7-4 で述べる S3 ストレージコストが増加する点はトレードオフとして把握しておく必要があります。

7-3. CloudWatch 監視メトリクスとアラーム設計

配信先に紐づくメトリクスにはアラームを設定することが公式に推奨されています。本記事の構成で特に重要なメトリクスは以下の通りです。

アラーム設計の勘所として、DeliveryToS3.DataFreshness がバッファ間隔設定値の 2〜3 倍を継続して超える場合は配信遅延として通知し、ExecuteProcessing.Success や FailedConversion.Records は変換ロジックやスキーマ不整合の早期検知に使います。これらのメトリクスは 1 分間隔で収集され、DeliveryStreamName ディメンションを使って配信ストリームごとに絞り込めます。

DeliveryToS3.DataFreshness に対するアラームは、AWS CLI からは以下のように作成できます。

aws cloudwatch put-metric-alarm \
  --alarm-name "firehose-delivery-freshness-high" \
  --namespace "AWS/Firehose" \
  --metric-name "DeliveryToS3.DataFreshness" \
  --dimensions Name=DeliveryStreamName,Value=prod-clickstream-firehose \
  --statistic Maximum \
  --period 300 \
  --evaluation-periods 3 \
  --threshold 1800 \
  --comparison-operator GreaterThanThreshold \
  --alarm-actions arn:aws:sns:ap-northeast-1:123456789012:firehose-alerts

上記は「直近 5 分間の最大値が 1,800 秒(30 分)を 3 回連続で超えたら通知する」設定です。バッファ間隔を 900 秒に設定しているストリームであれば、その 2 倍を閾値に取ることで、リトライの累積による配信遅延を早期に検知できます。

7-4. コスト最適化の勘所

Firehose のコストは主に以下の要素で構成されます(料金は変更される可能性があるため、実際の設計時は最新の料金ページで確認してください)。

• 取り込みデータ量課金: Direct PUT / Kinesis Data Streams ソースは 5 KB 単位で切り上げ課金されます。3 KB のレコードでも 5 KB 分課金されるため、極端に小さいレコードを大量に送るより、送信側でまとめてバッチ化するとコスト効率が良くなります。
• Format Conversion 課金: 変換対象データ量(GB)に応じて課金されます。ORC/Parquet 変換はクエリコスト削減効果が大きい一方、変換自体にも課金が発生するため、Athena/Redshift Spectrum 側のスキャン削減効果とのバランスで要否を判断します。
• 動的パーティショニング課金: 処理データ量(GB)に加え、配信される S3 オブジェクト数、jq 処理を使う場合は JQ 処理時間にも課金が発生します。パーティションキーを細かくしすぎるとオブジェクト数課金と S3 PUT リクエスト数が増加するため、7-1 で述べたバッファ間隔とのトレードオフで粒度を決めます。
• Lambda 変換課金: Firehose 自体の課金とは別に、Lambda の実行時間・呼び出し回数に応じた課金が発生します。バッファサイズを大きくすると呼び出し回数は減りますが、1 回あたりの実行時間が伸びるため、6 章のペイロード上限(6 MB)とあわせて総コストを見積もる必要があります。

7-5. トラブルシューティング — 詰まりやすいポイントの切り分け

本番運用で遭遇しやすい事象と、切り分けに使うメトリクス・設定の組み合わせを整理します。

いずれの事象も、まず ErrorOutputPrefix 配下に退避されたレコードとエラーメッセージ(errorCode / errorMessage)を確認し、次に該当する CloudWatch メトリクスで発生タイミングと規模を把握する、という順序で切り分けると原因特定が速くなります。

8. まとめ

8-1. まとめ

1-1で述べたとおり、本記事は「配信ストリームの設計原則から、宛先の選定基準、Iceberg/Snowflakeという新宛先の実践的な位置づけまでを一気通貫で整理する」ことをゴールに掲げていました。§2で改称の経緯と配信ストリームの基本構造を、§3〜§4でソース設定と既存宛先を、§5で本記事の核となるIceberg/Snowflake宛先を、§6〜§7でLambda変換とバッファリング/監視/コスト最適化を、それぞれ実装レベルで積み上げてきました。振り返ると、本記事の要点は次の3点に集約されます。

①Firehose自体を主役に据えた整理

既存記事群では、Firehose はデータ分析基盤の一要素、あるいはログ配信経路の一つとして局所的に言及されるにとどまっていました。本記事は、Direct PUT / Kinesis Data Streams / Amazon MSK というソースの違い、S3 / Redshift / OpenSearch Service / Splunk という既存宛先の選定基準、Lambda 変換とバッファリング設計までを、Firehose という配信ストリームそのものを主題として整理した点が最大の特徴です。

②Iceberg/Snowflakeという新宛先の独自性

2024年10月に追加された Apache Iceberg テーブル宛先と、コンソールから直接設定できる Snowflake 宛先は、既存コーパス内のどの記事にも言及がない完全な空白領域でした。§5 で扱った UniqueKeys の設計・JSON式ルーティングによる複数テーブル振り分け・東京リージョンでの対応状況は、いずれも本記事が独自に整理した内容です。

③本番運用としての深度

機能紹介にとどまらず、動的パーティショニングのアクティブパーティション上限(500)・Lambda変換のペイロード上限(6MB)・バッファサイズ/インターバルのトレードオフ(1–128MB・60–900秒)・CloudWatchメトリクスによるアラーム設計・コスト構成という、実際に本番投入する際に判断が必要になる項目を一貫して扱いました。

これら3点は独立した論点ではなく、相互に関連しています。Firehoseを主役として整理したからこそ(①)、既存記事群では扱われてこなかったIceberg/Snowflake宛先の実装レベルの詳細(②)を掘り下げることができ、その掘り下げの過程で、動的パーティショニングやLambda変換の上限値といった本番運用上の落とし穴(③)が明らかになりました。1-3で述べた「Firehoseを主役に据えた記事の不在」という空白は、単に取り上げる記事がなかったというだけでなく、こうした本番運用の深度で語られる機会そのものがなかったことを意味しています。

本記事を読み終えた読者が「次に何をすべきか」を読者層別に整理すると、次のとおりです。

これらの整理を踏まえ、実際にどの宛先を選ぶべきか迷った際のクイックリファレンスとして、次の表を活用してください。

本記事全体を通じて登場した主要な数値・上限値を、設計時のリファレンスとして一覧化しておきます。個別の章で詳細を確認する際の索引としても活用してください。

8-2. よくある選定ミス

Amazon Data Firehose の設計・運用でつまずきやすいポイントを、本記事で扱った内容に沿って整理しました。導入前・実装前のセルフチェックリストとして活用してください。

以下の8選は、§2〜§7で扱った実装レベルの詳細のうち、特に「一見動いているように見えるが、データ量やリクエストパターンが変化した際に初めて問題化する」性質を持つものを中心に選定しています。単発の設定ミスというより、設計判断の前提となる公式仕様の理解不足が背景にある点が共通しています。

選定ミスの共通パターンと根本原因

上記8選を俯瞰すると、3つの共通パターンに帰着します。

1. 改称・機能追加への追随不足(ミス1・4): 2024年の改称やIceberg/Snowflake宛先追加のような変更に対し、公式ドキュメントを都度確認せず、過去に得た理解のまま設計してしまうことが原因です。特にリージョン対応状況のような「後から変わりうる」情報は、実装直前の再確認を運用ルールに組み込む必要があります。

2. 上限値の見落とし(ミス2・3・6・7): バッファサイズ・Lambdaペイロード・アクティブパーティション数・MSKレコードサイズなど、Firehoseには複数のレイヤーで異なる上限値が存在します。それぞれの上限がどの設定(ソース種別・変換の有無)に依存して変わるのかを事前に整理せず、本番投入後にエラー出力先の増加で初めて気づくケースが多く見られます。

3. テーブル/スキーマ設計とFirehose設定の不整合(ミス5・8): IcebergテーブルのUniqueKeysやGlueスキーマ定義は、Firehose側の設定ではなくテーブル・カタログ側の設計事項ですが、Firehoseの配信ストリーム設計と独立に検討されがちです。宛先テーブルの設計とFirehose側の変換・振り分け設定は、必ずセットで検証してください。

これら3パターンに共通するのは、いずれも「本番投入前には気づきにくく、データ量が増えた後や機能追加後に顕在化する」という点です。開発環境での動作確認だけでは、バッファ上限やパーティション上限、リージョン対応状況の変化といった問題は表面化しません。§7-3で扱ったCloudWatchメトリクス(PartitionCountExceeded・ExecuteProcessing.Success・FailedConversion.Records等)へのアラーム設定を、本番投入時のチェックリストに組み込んでおくことが、これらのミスを早期に検知する最も確実な方法です。

8-3. 参考文献

本記事の内容は、Amazon Data Firehose の公式ドキュメントおよび公式発表に基づいて整理しています。より詳細な仕様やAPIリファレンス、最新のリージョン対応状況を確認する際は、公式ドキュメントを一次情報源として参照してください。本記事で扱った改称・新宛先追加の経緯を直接確認したい場合は、以下の公式発表も参考になります。

• Amazon Data Firehose Developer Guide(サービス概要・API仕様・クォータの一次情報源)
• Apache Iceberg tables 宛先GA発表(2024-10)・対応リージョン拡大発表(2024-11)
• Firehose クォータページ(バッファサイズ・インターバル・Lambda変換ペイロード上限などの最新値)

いずれも変更される可能性があるため、本番導入の直前には必ず最新版を確認する運用を組み込んでください。特に本記事で扱った改称の非対称性(§2-1)・バッファ上限値(§7-1)・Lambda変換のペイロード上限(§6-1)・Iceberg/Snowflake宛先のリージョン対応状況(§5-4)は、いずれも公式ドキュメントの該当ページを直接参照して記載内容を検証した項目です。第三者の解説記事やブログではなく、一次情報源であるAWS公式ドキュメントを起点に設計判断を行う習慣は、Firehoseに限らずAWSサービス全般の本番運用において引き続き有効です。

Amazon Data Firehoseは、改称とIceberg/Snowflake宛先の追加を経て、単なる「S3への中継サービス」から「データレイクへの構造化された配信レイヤー」へと役割を広げつつあります。本記事で整理したソース設定・宛先選定・変換・バッファリング設計・よくある選定ミスを踏まえ、自組織の配信ストリームを本番運用の水準まで引き上げる際の判断材料として活用してください。

AWS公式 Amazon Data Firehose Developer Guide を見る →