developersIO
Kinesis Data Streams が S3 汎用バケットへの直接配信をサポートしたので試してみた

Kinesis Data Streams が S3 汎用バケットへの直接配信をサポートしたので試してみた

Kinesis Data Streams から S3 の汎用バケットへ、Firehose を挟まず直接配信できるようになりました。On-Demand Standard のストリームから GZIP / JSON で配信し、Hive 形式のキーで届くこと、オブジェクトの中身、配信までにかかった時間、不正なレコードの行き先を確認しました。
Kinesis Data StreamsAmazon S3アップデートAmazon Data FirehoseAWS
2026.09.02

はじめに

2026年8月のアップデートで、Kinesis Data Streams から S3 の汎用バケットへ直接データを配信できるようになりました。これまで S3 へアーカイブするには Amazon Data Firehose を経由するのが一般的でしたが、サービス実行ロールを用意し、ストリームにチャネルを1つ作るだけで配信できます。

https://aws.amazon.com/about-aws/whats-new/2026/08/kinesis/data-delivery-general-purpose-s3-buckets

従来の構成と直接配信を並べると、次のようになります。

項目 KDS → Firehose → S3 KDS → S3 直接配信
追加で作るリソース Firehose ストリーム チャネル
データ変換・形式変換 できる できない
最小バッファ 60 秒（ゼロバッファリングも選択可） 300 秒
配信に伴う追加料金（単価） data-out $0.040/GB ＋ ingestion $0.029/GB S3 delivery $0.0275/GB

直接配信の料金は、既存の KDS 料金（per-stream / data-in / retention）に追加でかかります。Firehose の ingestion は 5KB 単位に切り上げて課金され、$0.029/GB は月間の最初の 500 TB までに適用される単価です。単価は Kinesis Data Streams の料金（S3 delivery タブ）と Amazon Data Firehose の料金に載っている US East の計算例の値です。最小バッファは、直接配信が配信のクォータと制約、Firehose がバッファリングのヒントに記載の値です。

この記事では、On-Demand Standard のストリームから SSE-S3 のバケットへ GZIP / JSON で配信する構成を作ります。配信の作成手順、S3 のキー構造、配信されたオブジェクトの中身、最初のレコードが届くまでの経過時間、不正なレコードの行き先を確認します。

検証内容

ap-northeast-1 で、On-Demand Standard のストリームにチャネルを作成しました。

検証環境

操作はすべて AWS CLI で実行しました。使ったバージョンは次のとおりです。

aws-cli/2.36.36 Python/3.14.6 Linux/7.1.6-400.asahi.fc44.aarch64+16k docker/aarch64.amzn.2023

2.36.36 では kinesis create-channel を実行できました。コマンドが見つからない場合は CLI を更新します。

実行環境は Docker イメージ public.ecr.aws/aws-cli/aws-cli:2.36.36 です。digest は sha256:7047de9ebb37c39d07f652ff2ff3f6c42308a4e189788aeeae70563b0c82b57d です。

送信元ストリームの設定は次のとおりです。

{
    "StreamDescriptionSummary": {
        "StreamName": "kds-s3-gp-ods-20260901-2210",
        "StreamARN": "arn:aws:kinesis:ap-northeast-1:123456789012:stream/kds-s3-gp-ods-20260901-2210",
        "StreamStatus": "ACTIVE",
        "StreamModeDetails": {
            "StreamMode": "ON_DEMAND"
        },
        "RetentionPeriodHours": 24,
        "StreamCreationTimestamp": "2026-09-01T13:58:21+00:00",
        "EncryptionType": "NONE",
        "OpenShardCount": 4,
        "ConsumerCount": 0,
        "MaxRecordSizeInKiB": 1024,
        "ChannelCount": 0
    }
}

直接配信の対象になるのは On-Demand のストリームです。

配信の作成

先にサービス実行ロールを作ります。信頼する principal は kinesis.amazonaws.com で、送信元アカウントと、チャネル ARN のワイルドカードを条件に付けました。

{
    "Role": {
        "Path": "/",
        "RoleName": "kds-s3-gp-role-20260901-2210",
        "Arn": "arn:aws:iam::123456789012:role/kds-s3-gp-role-20260901-2210",
        "CreateDate": "2026-09-01T13:58:07+00:00",
        "AssumeRolePolicyDocument": {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Principal": {
                        "Service": "kinesis.amazonaws.com"
                    },
                    "Action": "sts:AssumeRole",
                    "Condition": {
                        "StringEquals": {
                            "aws:SourceAccount": "123456789012"
                        },
                        "ArnLike": {
                            "aws:SourceArn": "arn:aws:kinesis:ap-northeast-1:123456789012:channel/*"
                        }
                    }
                }
            ]
        }
    }
}

権限ポリシーには、配信先バケットへの書き込みとバケット一覧の取得、DLQ 用のアクションを入れました。この構成で配信が成功しました。マルチパートアップロードの実行自体は s3:PutObject で許可されます（CreateMultipartUpload）。

サービス実行ロールの権限ポリシー
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "DeliveryBucketList",
            "Effect": "Allow",
            "Action": [
                "s3:ListBucket",
                "s3:ListBucketMultipartUploads"
            ],
            "Resource": [
                "arn:aws:s3:::kds-s3-gp-20260901-2210-123456789012",
                "arn:aws:s3:::kds-s3-gp-20260901-2210-123456789012/*"
            ]
        },
        {
            "Sid": "DeliveryBucketWrite",
            "Effect": "Allow",
            "Action": [
                "s3:PutObject",
                "s3:CreateMultipartUpload",
                "s3:UploadPart",
                "s3:CompleteMultipartUpload",
                "s3:ListMultipartUploads",
                "s3:ListMultipartUploadParts"
            ],
            "Resource": [
                "arn:aws:s3:::kds-s3-gp-20260901-2210-123456789012/*"
            ]
        },
        {
            "Sid": "DLQBucketAccess",
            "Effect": "Allow",
            "Action": [
                "s3:PutObject",
                "s3:ListBucket",
                "s3:ListBucketMultipartUploads"
            ],
            "Resource": [
                "arn:aws:s3:::kds-s3-gp-20260901-2210-123456789012",
                "arn:aws:s3:::kds-s3-gp-20260901-2210-123456789012/*"
            ],
            "Condition": {
                "StringEquals": {
                    "aws:ResourceAccount": "123456789012"
                }
            }
        }
    ]
}

配信の設定は入力 JSON にまとめて渡しました。

{
  "ChannelName": "kds-s3-gp-channel-20260901-2210",
  "ServiceExecutionRoleARN": "arn:aws:iam::123456789012:role/kds-s3-gp-role-20260901-2210",
  "StreamConfigurationList": [
    {
      "StreamARN": "arn:aws:kinesis:ap-northeast-1:123456789012:stream/kds-s3-gp-ods-20260901-2210",
      "RecordConfiguration": { "RecordFormatType": "JSON" }
    }
  ],
  "S3DestinationConfiguration": {
    "DataFreshnessInSeconds": 300,
    "DeadLetterQueueS3Configuration": {
      "BucketARN": "arn:aws:s3:::kds-s3-gp-20260901-2210-123456789012",
      "ExpectedBucketOwner": "123456789012",
      "ErrorOutputPrefix": "dlq/"
    },
    "StorageConfiguration": {
      "BucketARN": "arn:aws:s3:::kds-s3-gp-20260901-2210-123456789012",
      "ExpectedBucketOwner": "123456789012",
      "OutputKeyTemplate": "data/year=!{yyyy}/month=!{MM}/day=!{dd}/hour=!{HH}/records-!{extension}",
      "StorageClass": "STANDARD",
      "CompressionType": "GZIP"
    }
  }
}

aws kinesis create-channel \
  --cli-input-json file://scripts/params/create-channel.json

DLQ の設定は S3DestinationConfiguration の直下にあり、StorageConfiguration の中ではありません。また、圧縮に GZIP を指定する場合は、キーテンプレートに !{extension} が必要です。暗号化を指定していないので、配信先は S3 既定の SSE-S3 です。

create-channel の応答
{
    "ChannelDescription": {
        "ChannelName": "kds-s3-gp-channel-20260901-2210",
        "ChannelARN": "arn:aws:kinesis:ap-northeast-1:123456789012:channel/xxxxxxxxxxxxxxxxx",
        "ChannelId": "xxxxxxxxxxxxxxxxx",
        "ChannelStatus": "CREATING",
        "ChannelCreationTimestamp": "2026-09-01T13:58:38+00:00",
        "ServiceExecutionRoleARN": "arn:aws:iam::123456789012:role/kds-s3-gp-role-20260901-2210",
        "StreamConfigurationList": [
            {
                "StreamARN": "arn:aws:kinesis:ap-northeast-1:123456789012:stream/kds-s3-gp-ods-20260901-2210",
                "StreamCreationTimestamp": "2026-09-01T13:58:21+00:00",
                "RecordConfiguration": {
                    "RecordFormatType": "JSON"
                }
            }
        ],
        "S3DestinationConfiguration": {
            "DataFreshnessInSeconds": 300,
            "DeadLetterQueueS3Configuration": {
                "BucketARN": "arn:aws:s3:::kds-s3-gp-20260901-2210-123456789012",
                "ExpectedBucketOwner": "123456789012",
                "ErrorOutputPrefix": "dlq/"
            },
            "StorageConfiguration": {
                "BucketARN": "arn:aws:s3:::kds-s3-gp-20260901-2210-123456789012",
                "ExpectedBucketOwner": "123456789012",
                "OutputKeyTemplate": "data/year=!{yyyy}/month=!{MM}/day=!{dd}/hour=!{HH}/records-!{extension}",
                "StorageClass": "STANDARD",
                "CompressionType": "GZIP"
            }
        },
        "LoggingConfiguration": {
            "CloudWatchLogs": {
                "Enabled": false
            }
        }
    }
}

作成直後のステータスは CREATING でした。少し待って取得した応答から、ステータスの部分を抜き出しました。

{
    "ChannelDescription": {
        "ChannelName": "kds-s3-gp-channel-20260901-2210",
        "ChannelStatus": "ACTIVE",
        "ChannelCreationTimestamp": "2026-09-01T13:58:38+00:00"
    }
}

作成から ACTIVE の観測までは、13:58:38Z から 13:58:46Z までの 8 秒でした。create-channel の応答には、指定したレコード形式・GZIP 圧縮・保存先と DLQ の設定がそのまま反映されていました。

キー構造

{
    "Contents": [
        {
            "Key": "data/year=2026/month=09/day=01/hour=13/records--000000000000000000-000000-9d6757de83b.gz",
            "LastModified": "2026-09-01T14:15:52+00:00",
            "ETag": "\"dd02b756664cd84f8b3d675727905c1d-1\"",
            "ChecksumAlgorithm": [
                "SHA256"
            ],
            "ChecksumType": "COMPOSITE",
            "Size": 66,
            "StorageClass": "STANDARD"
        },
        {
            "Key": "data/year=2026/month=09/day=01/hour=13/records--000000000000100000-000000-9d6757de83b.gz",
            "LastModified": "2026-09-01T14:16:17+00:00",
            "ETag": "\"2b73988bd978d1be87230ceb356aff6e-1\"",
            "ChecksumAlgorithm": [
                "SHA256"
            ],
            "ChecksumType": "COMPOSITE",
            "Size": 89,
            "StorageClass": "STANDARD"
        }
    ],
    "RequestCharged": null,
    "Prefix": "data/"
}

指定したキーテンプレートのとおり、年・月・日・時を Hive 形式のパーティションとして含むキーになりました。テンプレートで指定したのは records- までで、その後ろの連番とユニークな接尾辞、.gz はサービスが付けます。有効な JSON のレコード（次の節で投入した 3 件）は 2 つのオブジェクトに分かれ、連番を含むキーで区別されました。

配信オブジェクトの中身

有効な JSON 3 件と、JSON ではない文字列 1 件を投入しました。beta のレコードだけ、コロンとカンマの後に空白を入れています。

aws kinesis put-record --stream-name "$STREAM_NAME" --partition-key alpha \
  --cli-binary-format raw-in-base64-out \
  --data '{"message":"alpha","n":1,"nested":{"z":true}}'
aws kinesis put-record --stream-name "$STREAM_NAME" --partition-key beta \
  --cli-binary-format raw-in-base64-out \
  --data '{"message": "beta", "n": 2, "array": [1, 2, 3]}'
aws kinesis put-record --stream-name "$STREAM_NAME" --partition-key gamma \
  --cli-binary-format raw-in-base64-out \
  --data '{"message":"gamma","n":3,"null":null}'
aws kinesis put-record --stream-name "$STREAM_NAME" --partition-key invalid \
  --cli-binary-format raw-in-base64-out \
  --data 'not-json'

配信された 2 つのオブジェクトを GZIP 解凍して連結すると、こうなりました。1 つ目（連番 000000000000000000）に beta、2 つ目に alpha と gamma が入っています。

{"message": "beta", "n": 2, "array": [1, 2, 3]}
{"message":"alpha","n":1,"nested":{"z":true}}
{"message":"gamma","n":3,"null":null}

区切り文字を確かめるため、alpha と gamma が入ったオブジェクトを od -c で表示しました。

0000000   {   "   m   e   s   s   a   g   e   "   :   "   a   l   p   h
0000020   a   "   ,   "   n   "   :   1   ,   "   n   e   s   t   e   d
0000040   "   :   {   "   z   "   :   t   r   u   e   }   }  \r  \n   {
0000060   "   m   e   s   s   a   g   e   "   :   "   g   a   m   m   a
0000100   "   ,   "   n   "   :   3   ,   "   n   u   l   l   "   :   n
0000120   u   l   l   }  \r  \n
0000126

今回のオブジェクトでは、レコードの区切りは CRLF で、各レコードの末尾にも CRLF が付いていました。レコード形式に JSON を指定していても、今回の入力では空白の除去やキー順の変更といった再シリアライズは起きず、投入時の表記の差がそのまま保持されました。

配信までの経過時間

チャネルの ACTIVE を観測した時刻、最初のレコードの投入時刻、最初のオブジェクトの LastModified を並べました。

channel_active_observed_utc=2026-09-01T13:58:46+00:00
first_record_put_utc=2026-09-01T13:58:46+00:00
s3_first_object_last_modified_utc=2026-09-01T14:15:52+00:00
4 レコードそれぞれの投入時刻
record_1_put_utc=2026-09-01T13:58:46+00:00
record_2_put_utc=2026-09-01T13:58:47+00:00
record_3_put_utc=2026-09-01T13:58:48+00:00
record_4_put_utc=2026-09-01T13:58:49+00:00

1 つのチャネルで 1 回実行した今回の結果では、最初の PutRecord から最初のオブジェクトまでは約 1,026 秒、およそ 17 分でした。最初の PutRecord は ACTIVE を観測したのと同じ 13:58:46Z なので、チャネルの起動待ち時間を含む値ではありません。DataFreshnessInSeconds には最小値の 300 を設定しましたが、この値どおりのタイミングでは配信されませんでした。2 つ目のオブジェクトは、1 つ目の 25 秒後に作成されました。

DLQ

{
    "Contents": [
        {
            "Key": "dlq/DESERIALIZATION_ERROR/2026/09/01/13/kds-s3-gp-channel-20260901-2210-000000000000000000-000000-9d6757de83b.json",
            "LastModified": "2026-09-01T14:15:52+00:00",
            "Size": 372,
            "StorageClass": "STANDARD"
        }
    ],
    "RequestCharged": null,
    "Prefix": "dlq/"
}

指定した ErrorOutputPrefix の下に、エラー種別と時刻のパーティションが自動で付きました。オブジェクトの中身は 1 行の JSON です。

{"approximateArrivalTimestamp":1788271129616,"streamArn":"arn:aws:kinesis:ap-northeast-1:123456789012:stream/kds-s3-gp-ods-20260901-2210","shardId":"shardId-xxxxxxxxxxxxxxxxx","sequenceNumber":"xxxxxxxxxxxxxxxxx","errorCode":"Record.InvalidJson","errorMessage":"The record contains invalid JSON. Please verify the record is valid JSON."}

今回確認できたオブジェクトには、stream ARN・シャードID・シーケンス番号・到着時刻・エラー内容が JSON で出力され、レコード本体は含まれていませんでした。DLQ の内容だけでは、元のデータを復元できません。

S3 直接配信を利用できる条件と制約

Kinesis Data Streams から S3 へ直接配信するには、次の条件と制約があります。開発者ガイドと配信のクォータと制約を確認し、手元のストリームが該当するかを先に確認してください。

  • 送信元ストリームが AWS マネージドキー（aws/kinesis）で暗号化されている場合は使えません。CMK への移行が前提になります
  • キャパシティモードは On-Demand だけが対象で、Provisioned のストリームは対象外です
  • 作成後に更新できるのは DataFreshnessInSeconds と LoggingConfiguration だけです。圧縮・キーテンプレート・レコード形式・宛先バケットは変更できないため、変えるには削除して作り直します。既存データのバックフィルもありません
  • 1 つのストリームに作れる配信は 2 つまでで、Iceberg 用が 1 つ、汎用 S3 用が 1 つという内訳です。緩和申請はできません
  • キーテンプレートで使える変数は時刻・チャネル名・ストリーム名だけです。レコードの内容によるパーティショニングはできません
  • クロスリージョン配信はできません。宛先バケットだけは別アカウントを指定できます

Firehose との棲み分け

直接配信を利用できる場合でも、Firehose 側にしかない機能があります。Lambda によるデータ変換、Parquet / ORC への形式変換、レコード内容による動的パーティショニング、60 秒未満の鮮度、1 つのソースから複数の宛先への配信です。逆に直接配信側にしかないものとして、ZSTD 圧縮と、配信先ストレージクラスの指定（STANDARD / INTELLIGENT_TIERING / GLACIER_IR）があります。

まとめ

Kinesis Data Streams から S3 の汎用バケットへ、Firehose を介さず直接配信できました。検証では、Hive 形式のキーでの出力、JSON レコードの配信、無効な JSON レコードの DLQ 出力を確認しました。
Firehose の機能や Lambda による変換を必要としない場合は、今回追加された直接配信を試してみてください。

また、再処理やテストを見据えたアーカイブが目的なら、Kinesis Data Streams の保持期間を延長するよりも、S3 を再処理の起点として設計したほうが扱いやすい場面があります。データの長期保持を目的とする場合にも、直接配信を評価してみてください。

この記事をシェアする

AWSのお困り事はクラスメソッドへ

関連記事

DynamoDB Streamsを使ってS3にデータを自動でアーカイブしてみる
yoshii-hiroki
2026.08.01
Amazon Data Firehose データストリームを Apache Iceberg 形式のテーブル配信機能を試す！
石川覚
2024.10.04
Aurora DSQL CDC を動かしてみる
中村 修太
2026.05.25
[アップデート] Amazon Kinesis Data Streams のレコードサイズ上限が10倍に拡大されました
若槻龍太
2025.11.01