Amazon Kinesis と Data Firehose でストリームを S3 にためる

前提条件

1. リージョン: ap-northeast-1（東京）
2. AWS マネジメントコンソールにログイン済みで、Amazon Data Firehose・Amazon S3・IAM ロールの作成・AWS CloudShell を操作できる IAM 権限があること
3. 事前に作成しておくリソースはありません

作成するリソース一覧

配信先の S3 バケットを作成する

1. AWS マネジメントコンソールにログインし、リージョンが ap-northeast-1（東京）であることを確認します
2. 上部の検索バーに S3 と入力し、表示された「S3」を選択します
3. 左側ナビゲーションの「汎用バケット」を選択し、「バケットを作成」をクリックします
4. 以下を設定します
   • バケット名: handson-firehose-<アカウントID>（例: handson-firehose-123456789012）
5. その他の項目はデフォルトのままにします
6. ページ下部の「バケットを作成」をクリックします

バケット一覧に handson-firehose-<アカウントID> が表示されれば作成完了です。S3 の基本操作は「Amazon S3 でバケットを作成しオブジェクトを管理する」で扱っています。

Firehose ストリームを作成する

1. 上部の検索バーに Firehose と入力し、表示された「Amazon Data Firehose」を選択します
2. 「Firehose ストリームを作成」をクリックします
3. 「ソースと送信先を選択」で以下を設定します
   • ソース: Direct PUT
   • 送信先: Amazon S3
4. 「Firehose ストリーム名」に handson-firehose-to-s3 と入力します

1. 「レコードを変換および転換 - オプション」セクションは、「データ変換をオンにする」「レコード形式の変換を有効にする」「解凍をオンにする」のいずれもオフ（デフォルト）のままにします
2. 「送信先の設定」セクションで以下を設定します
   • S3 バケット: 「参照」をクリックし、表示された「Amazon S3 のバケットを選択する」の検索ボックスに handson-firehose と入力して、handson-firehose-<アカウントID> を選択し「選択」をクリック
   • 改行の区切り文字: 有効（デフォルトは「有効ではありません」なので、「有効」に切り替えます）
3. 「バッファのヒント、圧縮、ファイル拡張子、暗号化」を展開し、以下を設定します
   • バッファサイズ: 5 MiB（デフォルトのまま）
   • バッファ間隔: 60 秒（結果を早く確認するため、デフォルトの 300 秒から短くします）
4. S3 バケットプレフィックス、圧縮、暗号化などその他の項目はデフォルトのままにします

1. 「詳細設定」を展開し、「サービスアクセス」が「IAM ロール KinesisFirehoseServiceRole-handson--ap-northeast-1-<数字> を作成または更新」になっている（デフォルト）ことを確認し、そのままにします
2. ページ下部の「Firehose ストリームを作成」をクリックします（直後に「Firehose ストリームが作成されませんでした」と表示された場合は、IAM ロールが反映されるまで数十秒待ってから、もう一度クリックします）
3. ストリームの詳細画面が表示され、ステータスが「作成中」から「アクティブ」に変わるまで待ちます

デモデータを送信する

1. handson-firehose-to-s3 の詳細画面で、「デモデータでテスト」セクションを展開します（折りたたまれています）
2. 「デモデータの送信を開始」をクリックします
3. 株価を模したデモデータ（{"TICKER_SYMBOL":"QXZ","SECTOR":"HEALTHCARE","CHANGE":-0.05,"PRICE":84.51} のような JSON）が Firehose ストリームに送信され始め、ボタンの表示が「デモデータの送信中」に変わります。送信はブラウザでスクリプトを実行して行うため、送信中はこの画面を開いたままにします（別のページに移動すると送信が止まります）

1. 2 分ほど待ったら、「デモデータの送信を停止」をクリックします

S3 にオブジェクトが配信されたことを確認する

1. 上部の検索バーに S3 と入力し、表示された「S3」を選択します
2. 「汎用バケット」から handson-firehose-<アカウントID> をクリックします
3. 「オブジェクト」タブに、年のフォルダ（例: 2026/）が作成されていることを確認します
4. 年/ → 月/ → 日/ → 時/ の順にフォルダをクリックして開きます
5. handson-firehose-to-s3-1-2026-... のような名前のオブジェクトが、1 つ以上作成されていることを確認します

時のフォルダは UTC で作られるため、日本時間とは 9 時間ずれます。オブジェクト名は <Firehose ストリーム名>-<Firehose ストリームバージョン>-<年>-<月>-<日>-<時>-<分>-<秒>-<UUID> の形式で、バッファ間隔の 60 秒ごとにまとめて配信されていることが確認できます。

CloudShell から自分のデータを送信する

次に、CloudShell から AWS CLI の put-record で、自分で作ったレコードを Direct PUT で送信します。

1. コンソール上部（またはコンソール左下）の CloudShell アイコンをクリックし、CloudShell を起動します
2. 以下のコマンドを実行し、3 件のレコードを送信します。AWS CLI バージョン 2 ではバイナリ型のパラメータに base64 文字列を渡すのが既定のため、--cli-binary-format raw-in-base64-out を付けて JSON 文字列をそのまま渡します

for i in 1 2 3; do
  aws firehose put-record \
    --delivery-stream-name handson-firehose-to-s3 \
    --record "{\"Data\":\"{\\\"tree\\\":\\\"banana-0${i}\\\",\\\"ripe\\\":${i}}\"}" \
    --cli-binary-format raw-in-base64-out
done

1. 各コマンドの出力に RecordId が表示されることを確認します

1. 60 秒（バッファ間隔）以上待ってから、以下のコマンドでバケット内のオブジェクトを一覧表示します

BUCKET=handson-firehose-$(aws sts get-caller-identity --query Account --output text)
aws s3 ls s3://${BUCKET}/ --recursive

1. 一覧の最後（最新）のオブジェクトのキーを指定して、中身を表示します。<オブジェクトキー> は一覧に表示されたキーに置き換えます。オブジェクトの末尾には改行が入らず、次のプロンプトが最後の行に重なって見えてしまうため、末尾に ; echo を付けて改行を出力します

aws s3 cp s3://${BUCKET}/<オブジェクトキー> - ; echo

1. 以下のように、送信した 3 件のレコードが 1 行ずつ改行で区切られて、1 つのオブジェクトにまとまっていることを確認します

{"tree":"banana-01","ripe":1}
{"tree":"banana-02","ripe":2}
{"tree":"banana-03","ripe":3}

1 件ずつ送ったレコードが、バッファ間隔のあいだにまとめられ、改行区切りの 1 つのオブジェクトとして S3 に配信されたことが確認できます。