DEA / ドメイン1 データの取り込みと変換
重要な内容
Amazon Kinesis Data Streams
は、クリックログ、IoT のセンサー値、アプリケーションのイベントなどを、順序を保ったままリアルタイムに受け取り、一定期間保持して複数の処理系に読ませるサービスです。DEA では、ストリーミングソースからの読み取り、 の呼び出し、への対処、再処理の設計の中心として出題されます。
Point
- Shard (シャード) は、ストリームの容量と順序の単位です。「かけら」の意味です。
- KPL = Kinesis Producer Library、KCL = Kinesis Client Library です。KPL は書き込み側、KCL は読み取り側のライブラリです。
- 和訳
- Shard: 破片、かけら
- Weblio英和辞書「shard」
概要
プロデューサーはレコードにを付けて書き込みます。キーのハッシュ値でが決まり、シャード内では書き込んだ順に並びます。コンシューマーはシャードごとにレコードを読み、どこまで読んだかを記録します。レコードは保持期間のあいだ残るため、複数のコンシューマーが独立して読み、必要なら読み直せます。
試験では、容量モードの選択、パーティションキーの設計、コンシューマーの方式 (共有か か)、 との連携の設定、の対処を判断します。
主要技術
①シャードとパーティションキー
| 項目 | 内容 |
|---|---|
| 書き込み上限 (あたり) | 毎秒 1 MB または毎秒 1,000 レコード |
| 読み取り上限 (シャードあたり、共有) | 毎秒 2 MB、毎秒5回の GetRecords |
| 1回の GetRecords | 最大 10,000 レコード、最大 10 MB |
| PutRecords の1リクエスト | 最大 500 レコード |
| 順序の保証 | 同じシャード内 (同じ) だけ |
パーティションキーは、順序が必要な単位で、かつ値の種類が十分に多いものを選びます。たとえば「全イベントで同じキー」にすると、1シャードに集中して他のシャードが遊びます (ホットシャード)。
②容量モード: プロビジョンドとオンデマンド ⚠️ 頻出
| モード | 容量の決め方 | 向くケース | 注意点 |
|---|---|---|---|
| プロビジョンド | 数を自分で決め、UpdateShardCount やシャードの分割・結合で増減する | 流量が予測でき、コストを細かく管理したい | ピークに合わせて増やす運用が必要 |
| 流量に合わせてサービスがシャードを自動で増減する。新規作成時の書き込みは毎秒 4 MB から始まる | 流量が読めない、急に増える、運用を減らしたい | 1つのへの集中は、どちらのモードでも1シャードの上限を超えられない |
モードの切り替えは、1ストリームにつき 24 時間に2回までです。
⚠️ 頻出: 「トラフィックが予測できず、シャードの管理をしたくない」はオンデマンドモードです。「特定のキーだけがされる」は容量モードではなくパーティションキーの問題です。
③プロデューサー
| 方法 | 特徴 |
|---|---|
| AWS (PutRecord / PutRecords) | シンプル。PutRecords でまとめて送ると呼び出し回数を減らせる |
| Producer Library (KPL) | 複数のユーザーレコードを1つの Kinesis レコードに集約 (aggregation) し、まとめて送る。再試行も行う |
| Kinesis エージェント | サーバーのログファイルを監視してストリームへ送る |
| AWS サービス連携 | Logs のサブスクリプション、 の Kinesis Data Streams 連携、 のターゲットなど |
PutRecords は一部のレコードだけが失敗することがあります。応答の FailedRecordCount を確かめ、失敗したレコードだけを再送します。
④コンシューマー: 共有スループットと Enhanced Fan-Out ⚠️ 頻出
| 方式 | 読み取りの帯域 | 配信方法 | 遅延の目安 |
|---|---|---|---|
| 共有 (標準) | あたり毎秒 2 MB を全コンシューマーで分け合う | コンシューマーが GetRecords で取りに行く (プル) | 読む処理系が増えるほど伸びる |
| (EFO) | 登録したコンシューマーごとに、シャードあたり毎秒 2 MB を専用で持つ | SubscribeToShard で /2 によりプッシュされる | コンシューマーの数によらず低い |
EFO のコンシューマーは、1ストリームあたり20個まで登録できます (On-demand Advantage モードでは50個まで)。EFO は読み取ったデータ量に応じた追加の料金がかかるため、コンシューマーが少なく遅延の要件が緩いなら共有スループットで十分です。
を使った自前のコンシューマーは、シャードの割り当てと (どこまで処理したか) を のテーブルで管理します。KCL のアプリケーションを複数台で動かすと、シャードをワーカー間で分担します。
⑤Lambda のイベントソースマッピング ⚠️ 頻出
は、イベントソースマッピングを使って のを読み、レコードのまとまり (バッチ) で関数を呼び出します。呼び出しは同期で、シャード内の順序を保ちます。
| 設定 | 範囲・既定 | 使いどころ |
|---|---|---|
| BatchSize | 既定 100、最大 10,000 | 1回の呼び出しで処理する件数 |
| MaximumBatchingWindowInSeconds | 0〜300 秒 | 少量のレコードで何度も呼ばれるのを防ぐ |
| ParallelizationFactor | 1〜10 (既定 1) | 1シャードを複数の呼び出しで並行処理する。単位の順序は保たれる |
| BisectBatchOnFunctionError | 既定 false | 失敗したバッチを二分割して再試行し、不正なレコードを絞り込む |
| MaximumRetryAttempts | -1 (無制限、既定)〜10,000 | 失敗時の再試行回数 |
| MaximumRecordAgeInSeconds | -1 (無制限、既定)〜604,800 | 古いレコードを諦める時間 |
| DestinationConfig (失敗時の送信先) | キューまたは トピック | 諦めたバッチの情報を退避する |
| FunctionResponseTypes = ReportBatchItemFailures | — | 失敗したレコードだけを報告し、成功分を再処理しない |
| StartingPosition | TRIM_HORIZON、LATEST、AT_TIMESTAMP | 読み始める位置 |
| TumblingWindowInSeconds | 0〜900 秒 | 短い時間窓の集計 |
既定では、関数がエラーを返すとそのバッチはレコードが期限切れになるまで再試行され、そのシャードの後続のレコードが止まります (ポイズンピル)。再試行回数の上限、レコードの最大経過時間、二分割、失敗時の送信先を設定して、詰まりを防ぎます。
イベントソースマッピングは同じレコードを重複して処理することがあるため、関数はに作ります。
⚠️ 頻出: 「1件の不正なレコードでシャードの処理が止まる」は、BisectBatchOnFunctionError、MaximumRetryAttempts、失敗時の送信先 (SQS) の組み合わせで解決します。「処理が追いつかず IteratorAge が伸びる」は、ParallelizationFactor を上げる、バッチサイズを見直す、シャードを増やす、EFO を使う、のいずれかです。
⑥スロットリングとホットシャード
| エラー・兆候 | 原因 | 対処 |
|---|---|---|
| 書き込みで ProvisionedThroughputExceededException | の書き込み上限を超えた | で再送、シャードを増やす、モード、キーの分散 |
| 一部のシャードだけ WriteProvisionedThroughputExceeded が多い | の偏り (ホットシャード) | 偏ったシャードを分割 (SplitShard)、キーにランダムな接尾辞を付ける |
| 読み取りで ReadProvisionedThroughputExceeded | コンシューマーが共有の読み取り上限を奪い合う | EFO を使う、GetRecords の頻度を下げる |
| GetRecords.IteratorAgeMilliseconds が増え続ける | コンシューマーの処理が書き込みに追いつかない | コンシューマーの並列度を上げる ( の並列化係数、 のワーカー) |
のシャード単位の詳細 (拡張) を有効にすると、どのシャードが偏っているかを確かめられます。
⑦保持期間と再処理
| 項目 | 内容 |
|---|---|
| 既定の保持期間 | 24 時間 |
| 最大の保持期間 | 8,760 時間 (365 日) |
| 再処理の方法 | 新しいコンシューマーを TRIM_HORIZON や AT_TIMESTAMP から読ませる |
24 時間を超えて保持すると追加の料金がかかります。長期の保存や、何か月も後の再処理が目的なら、 で へ保存しておく方が安価です。
⑧セキュリティ
| 項目 | 方法 |
|---|---|
| によるサーバー側暗号化 | |
| アクセス制御 | ポリシー (kinesis:PutRecord、kinesis:GetRecords など)、で他アカウントからのアクセス |
| プライベートな接続 | インターフェイス ()。aws:SourceVpce の条件で経路を限定 |
試験対策
①判断マトリクス
| 問題文のキーワード | 選ぶ設定・サービス |
|---|---|
| 流量が読めない、管理をしたくない | モード |
| 流量が安定、コストを細かく管理 | プロビジョンドモード |
| コンシューマーが多い、読み取りの遅延を下げたい | |
| 書き込みの | シャード追加、バックオフ付き再送、キーの分散 |
| 特定のシャードだけ過負荷 | の見直し、シャード分割 |
| の処理が追いつかない | ParallelizationFactor を上げる |
| 不正なレコードでシャードが止まる | 二分割、再試行の上限、失敗時の送信先 |
| バッチ内の一部だけ失敗 | ReportBatchItemFailures |
| 数日前のデータから処理し直す | 保持期間を延ばす + AT_TIMESTAMP |
| コードを書かずに へ保存 | をコンシューマーにする |
| 互換が必要 |
②よくある誤答
- 順序の保証をストリーム全体と考えること。内だけです。
- 読み取りの奪い合いに対してシャードを増やすこと。EFO で解決します。
- のエラーを放置すること。既定では再試行が無制限で、シャードが止まります。
- 長期保存のために保持期間を365日にすること。 への保存の方が安価です。