re:Invent 2025: Amazon Kinesis Data Streamsの内部アーキテクチャとコスト削減機能
はじめに
海外の様々な講演を日本語記事に書き起こすことで、隠れた良質な情報をもっと身近なものに。そんなコンセプトで進める本企画で今回取り上げるプレゼンテーションはこちら！
re:Invent 2025 の書き起こし記事については、こちらの Spreadsheet に情報をまとめています。合わせてご確認ください
📖 re:Invent 2025: AWS re:Invent 2025 - Amazon Kinesis Data Streams under the hood (ANT423)
この動画では、Amazon Kinesis Data Streamsの内部アーキテクチャと最新機能について詳しく解説しています。Prime Day時に毎秒8億700万レコードを処理した実績を持つこのサービスは、shardという単位で水平スケーリングを実現しています。新機能のOn-Demand Advantageでは、ウォームスループットの事前設定により約40%のコスト削減が可能になり、最大10メガバイトの大きなレコードもサポートされました。内部的には、shardの分割・統合による順序保証、3つのアベイラビリティゾーンでのデータ複製、Timer Serviceを使った分散システムの時刻同期など、高可用性と低レイテンシーを実現する工夫が施されています。また、Kinesis Client LibraryやKinesis Producer Libraryを使ったベストプラクティスとして、レコードのバッチ処理や集約による効率的なデータ送信方法も紹介されています。
※ こちらは既存の講演の内容を最大限維持しつつ自動生成した記事になります。誤字脱字や誤った内容が記載される可能性がありますのでご留意下さい。
本編
セッション紹介：Amazon Kinesis Data Streamsの内部構造を探る400レベルトーク
Amazon Kinesis Data Streams の内部について、これからお話しします。これは 400 レベルのトークで、ANT423 です。本日はご参加いただきありがとうございます。私は John Morkel で、Software Development のディレクターをしています。そして、皆さんのような素敵な方々とお話しするとき以外は、Amazon Kinesis Data Streams を担当するチームをリードしています。本日は同僚の Ali Alemi が参加しています。Ali は Principal Data Streaming Architect で、Amazon Kinesis Data Streams の詳細な部分について皆さんにご説明します。これから詳しく掘り下げていきたいと思います。
本日カバーする内容について、簡単に概要をお話しします。 手を挙げてもらいたいことがあります。これはトークの中でも皆さんが好きではない部分かもしれませんが、ここで Amazon Kinesis Data Streams を使ったことがある方はいますか？素晴らしい、多くの方がいますね。当然のことながら、皆さんはこれらの用語にかなり詳しいと思いますが、念のため、私は皆さんが同じ言語を話しているように、これらの概念を紹介するつもりです。ここ 1 ヶ月間にリリースした新機能について、私たちは非常に興奮しています。これらは顧客から多くのリクエストをいただいていた機能です。
ここでハイライトについて簡単に説明します。その後、実際のストリーミングアプリケーションをデモンストレーションするデモに入ります。これはライブデモ、というより、皆さん自身がダウンロードして実行できるデモです。これは Ali によって作成されました。最近リリースした新機能の一部をデモンストレーションする、実際のストリーミングアプリケーションについて説明します。その後、Ali に引き継ぎます。Ali は Amazon Kinesis Data Streams の内部について、本当に興味深い部分を説明します。
Amazon Kinesis Data Streamsとは：サーバーレスストリーミングサービスの基本と圧倒的なスケール
Amazon Kinesis Data Streams とは何ですか？ 多くの皆さんはすでにご存知だと思います。これは完全にマネージドされたサーバーレスのデータストリーミングサービスです。ストリーミングアプリケーションでは、リアルタイムの設定でデータを収集、処理、分析します。Amazon Kinesis Data Streams が優れている点は、サーバーレスであること、スケールできること、エラスティックであること、非常にセキュアであること、そして高可用性があることです。
平均的なストリーミングアプリケーションはこのような形です。このダイアグラムの左側には、ストリーミングデータの入力があります。これはクリックストリーム、IoT デバイス、モバイルアプリ、またはアプリケーションサーバーからのログかもしれません。あらゆる種類の異なるデータソースが、本番ネットワーク、インターネット、またはどこからでも来ることができます。これらすべてのデータは単一のポイントに集約されます。これらのデータソースを producer と呼びます。左側の producer は、データを Amazon Kinesis Data Streams に送信します。これが集約ポイントです。
それでは、これは様々なオプションのスイートによって消費されます。Managed Service for Apache Flink、Spark on EMR、Firehose、Amazon EC2、そしてカスタムコードを書いて処理したい場合は Lambda などのサービスがあります。これらのコンシューマーアプリケーションは、ストリーミングアプリケーションアーキテクチャの中核です。このデータを処理したら、データレイクまたは何らかの分析データベースなどのリポジトリに保存して、それを使用してビジネス価値を引き出すことができます。
もう一度手を挙げてください。今年の Prime Day で Amazon で何か買った人はいますか？何か買った人は、今年の Prime Day 中に Amazon Kinesis Data Streams をピーク時に 1 秒あたり 8 億 700 万レコードまで押し上げた人たちの一人です。このスケール感を理解するために、1 日に処理されるグローバルなクレジットカード取引は、このレートで約 3 秒で処理できます。これは Prime Day イベント中に Amazon Kinesis Data Streams を通じて処理される、かなり膨大な量の情報です。
毎日、Amazon Kinesis Data Streams 上に顧客が持っている約 340 万のストリームにわたって、約 70 ペタバイトのデータを処理しています。個々のストリームのスケールに少しズームインすると、顧客は 1 つの単一ストリームを通じて 1 秒あたり最大 10 ギガバイトのデータを送信しています。
それは大きな数字に聞こえるかもしれませんが、これは大文字の B のギガバイトです。つまり、1 秒ごとに 4K ムービー 1 本分くらいなので、かなり深刻なデータ量について話しています。ここからさらにズームインして、Kinesis Data Streams が動作できるこの非常に大規模な スケールから、単一の Kinesis データストリームのスケールへと移ります。個々のストリームでは、1 秒あたりギガバイト単位のデータ取り込み、コンシューマーへの 1 秒あたりギガバイト単位のデータファンアウト、そして 1 秒あたり最大数百万レコードを処理する可能性があります。これはすべてマルチテナントソリューション内で実現でき、異なるアプリケーションが単一のストリーム内でデータを生成および消費できます。
Kinesis Data Streamsの4つの柱とshardの仕組み
Kinesis Data Streams は、私たちが基盤となる AWS サービスと呼ぶものです。 この基盤は 4 つの柱によって支えられています。まず、Kinesis Data Streams は弾力的です。つまり、負荷に応じて自動的にスケールアップおよびスケールダウンします。使いやすいというのが、正直なところ Kinesis Data Streams の最大の利点の 1 つです。インフラストラクチャを自分で管理する必要がなく、パッチやネットワーク接続について心配する必要もありません。Kinesis Data Streams の使いやすさのおかげで、これらすべてがあなたのために処理されます。
信頼性が高いので、非常に低いレイテンシーと予測可能なレイテンシーで動作し、同じパフォーマンスを保ったまま、スケールアウトさせることができます。これはどのアプリケーションを構築する際にも、本当に堅牢な基盤となります。最後に、AWS サービスとして当然のことながら、高い可用性を備えています。私たちは複数のアベイラビリティーゾーンにサービスを構築し、内部に多くの冗長性を組み込んでいます。ストリーミングサービスとして最高レベルの可用性が得られることを保証でき、これらの複雑さについて考える必要はありません。
では、Kinesis データストリームの基本単位である shard について、本当に詳しく見ていきましょう。 shard はスケーリングの原子的な単位で、非常に明確に定義された制限があります。shard は 1 秒あたり 1 メガバイトの書き込みと 2 メガバイトの読み取りが可能です。1 秒あたり約 1000 レコードの書き込みと 2000 レコードの読み取りをサポートしています。これはすべてのストリーミングアプリケーションがスケールする際の基盤となります。
shard の使用方法について興味深い点の 1 つが、partition key です。partition key は、Kinesis Data Streams にデータを書き込む際に指定するものです。その後、ハッシュ化され、shard 全体に分散されます。partition key と shard 間のマッピングは、アプリケーションがスケールアップおよびスケールダウンするにつれて、自動的に管理されます。このレベルの間接参照により、Kinesis Data Streams を使用して水平スケーラブルなアプリケーションを構築できます。
この shard の概念を持つことには、メリットとデメリットがあります。 スケーリングが可能になり、容量を個別の単位で管理する便利な方法であるため、従量課金制で利用できます。使用している容量についてのレベルの基本的な弾力性が得られます。しかし、いくつかの課題があります。partition key と shard 間のマッピングがあり、時にはそれが一目瞭然ではありません。また、スケールバックする際の複雑さもあります。データが異なる shard に再分散される場合、どこに行くのか、どのように対処するのかという問題があります。Kinesis Data Streams がこれをどのようにサポートしているかについては、後で説明します。また、コンシューマー間で負荷が不均等になる可能性がある課題も発生する可能性があります。shard は構成要素として非常に便利ですが、万能ではなく、それらを有用にするために私たちが行った興味深いことがいくつかあります。
オンデマンドスケーリングとKinesis On-Demand Advantageの登場
Kinesis Data Streams について顧客から聞いた要望の 1 つは、容量管理があまり好きではないということです。 Serverless は、一般的に容量の抽象的な単位でプロビジョニングするという点で、本当に大きな役割を果たします。
ただし、オンデマンドスケーリングは、お客様から本当にリクエストされていた機能でした。約4年前に、Kinesis Data Streams のオンデマンド版をローンチしました。このバージョンでは、特定のストリームに送信されるロードの量を自動的に監視します。必要に応じてシャードを追加したり、シャードをマージしたりします。つまり、トラフィックスパイクが発生して、予期しないレコードの急増がストリームに入ってきた場合、新しいシャードを自動的に追加します。心配する必要はありません。それは透過的に起こるだけです。
お客様がリクエストしているもう1つのことは、Kinesis Data Streams にもっと大きなレコードを書き込む機能です。つい最近まで、レコードの上限は1メガバイトでした。お客様はそれを回避するためにいろいろと工夫する必要がありました。Kinesis Data Streams のオンデマンド版の本当のメリットは、ストリーミングアプリケーションでお客様に見られる最大値である約10ギガバイト毎秒まで、完全にハンズフリーの容量管理ができることです。このスケーリングは自動的に行われます。しきい値を設定する必要はありません。オンデマンド版の Kinesis Data Streams を使用するとすぐに実行されます。
手動でプロビジョニングしたり、元の Kinesis Data Streams プロビジョニングモデルでプロビジョニングされたシャードを手動で管理したりした場合と同じパフォーマンスエンベロープが得られます。オンデマンドでは、シャード時間や未使用のシャードについて心配する必要はありません。取り込んだデータに対してのみ支払うだけで、これはかなり魅力的な従量課金モデルです。ここで CloudWatch グラフを見せて、プロビジョニングモードまたはオンデマンドモードのいずれかを使用するとどのように見えるかを示します。ここでは、約5つのシャードがプロビジョニングされた状態で、約4メガバイト毎秒のロードを快適に処理しているワークロードが見えます。その後、トラフィックスパイクが発生します。
スロットルされたレコードの量、つまり上のグラフの上の線が約60パーセントまで跳ね上がるのが見えます。これは良くありません。データが失われています。この場合、オンデマンド容量を有効にしました。これは既存のストリームに対して実行できることです。システムが既存のシャード上のロードがプロビジョニングされたスループットを超えたことを自動的に検出したのが見えます。既存のシャードが分割され、新しいシャードがプロビジョニングされました。入力側のスロットルされたデータの量がゼロに低下します。これは非常にシンプルです。数分以内に、完全にハンズオフで、スケーリングの問題が解決されました。
あなたのようなビルダーが経験してきた、ストリーミングアプリケーションを構築することについて私たちが聞いた課題のいくつかは、スケール時に予測可能なパフォーマンスが必要だということです。特にスパイキーなワークロードでスケールアップおよびスケールダウンするときに、予測可能なコストを持つことができたらいいですね。パフォーマンスエンベロープはかなり一貫性があるべきです。これらはすべて、ストリーミングアプリケーションを構築しようとしている場合の考慮事項です。先月ローンチしたことの1つは Kinesis On-Demand Advantage です。このオンデマンドの新しいバージョンが提供する2つの主要な機能があります。1つ目は、ウォームスループットを指定する機能です。ロードの増加が事前にわかっている場合、自分自身をオーバープロビジョニングして、アイドル状態のシャード時間に対して潜在的に支払う代わりに、ピーク時にこの量のデータが入ってくると予想していると言うだけです。
それでは、必要な時に利用できるこの程度のスループットを確保しておきたいということを先にお伝えしておきます。使い方は本当にシンプルです。有効にするだけです。コマンドラインツールまたはコンソールを使ってストリームを作成することができます。希望するスループットを指定するだけです。または、既存のストリームに対してもこれを指定することができます。
Kinesis Data Streams の機能として、オンデマンドには大きな利点があり、通常のオンデマンド価格と比べて約 40% 安くなっています。この価格ポイントを実現できた理由は、コミットされたワークロードであれば、リアクティブスケーリングと従量課金モデルを使いたい理由がある顧客向けにこの価格設定をサポートできることを判断するための作業を行ったからです。ただし、一定の使用量にコミットできるのであれば、はるかに良い価格ポイントで提供することができます。このオンデマンド利点は本当に興味深いものなので、Kinesis Data Streams のドキュメントで確認することをお勧めします。
大きなレコードサポート：最大10メガバイトへの拡張とトークンバケット方式
次に、顧客が課題を抱えていたのは大きなレコードです。 1 メガバイトより大きなデータを取り込む必要がある場合、通常は Amazon S3 などの別のデータストアにそれを保存してから、そのレコードへのポインタをデータストリームに保存するということをしていました。その後、コンシューマーはこのポインタを見たときに、そのレコードを取得しに行く必要があることを理解していました。これにはいくつかの課題がありました。まず、コンシューマーに組み込む必要のある複雑な処理がたくさんありました。次に、特定のシャードを通過するデータの量が、実際には特定のコンシューマーが実行する必要のある作業の量を表していません。これらの大きなレコードはスケーリングシステムとバランシングシステムに対して不透明であるため、異なるコンシューマー間でかなりの不均衡が生じる可能性があり、コンシューマーのパフォーマンスにかなり大きな停滞が生じる可能性があります。
私たちが行ったのは、大きなレコードのサポートを追加することです。 最大レコードサイズを 10 倍に増やしたので、単一のシャードに最大 10 メガバイトのレコードを送信できるようになりました。有効にするのは本当に簡単です。ストリームの設定するだけで、使用するのに追加コストはありません。ここで説明する興味深い点の 1 つは、個別のシャードの 1 メガバイト書き込みと 2 メガバイト読み取りの制限でこれがどのように機能するかということです。これらのスループット制限がある中で、個別のシャードに 10 メガバイトのデータを書き込むことができるのはなぜでしょうか。それについては次で説明します。
個別のシャードのスロットリングを実際に行う方法は、 トークンバケットと呼ばれるものを使用することです。トークンのバケットを保持していて、1 秒あたりのスループットを使用するときに、使用したトークンの量をそのバケットから差し引き、一定の速度で補充します。この場合、これらのバケットは 1 秒あたり約 1 メガバイトの速度で補充されます。大きなレコードが通過して 1 つの大きなバースト内で消費されるとき、利用可能なトークンがすべて消費されます。 この場合、このバケットには 1010 個のトークンがありました。これは問題ありません。バースト性とモデル内のマイクロバースト許容量があるため、シャードのハードプロビジョニングに対してまだ保持しているため、正常に通過することができました。ここで、バースト容量を使い果たしたことが見え、1 秒あたり 1 メガバイトのレベルで書き込みを続けることができます。
では、別のバーストが来たらどうなるでしょうか？すでに持っている毎秒1メガバイトを超えて利用可能なトークンがバケットにないため、スループットを少し落とすまでスロットルされることになります。この場合、スループットを約毎秒0.5メガバイトまで落としたのが見えますね。そしてそのトークンバケットは毎秒約0.5メガバイトの速度で補充されます。これは large record フィーチャーがどのように機能するかについて考える際の微妙な点ですが、これはかなりシームレスな使い方ができる方法だと思っています。
ライブデモ：ソーシャルメディアセンチメント分析でウォームスループットの威力を実証
Ali と私がこのトークのために一緒に作業していた時に、かなり面白いと思ったことの一つは、私たちが両方とも双子の親だったということです。Kinesis Data Streams のような水平スケーラブルなサービスに取り組みながら、双子の親であるというのは興味深いことです。
双子の親として、私たちは自分たち自身を水平にスケールする方法を学ぶ必要がありました。同時に、私たちの子どもたちも水平にスケールしています。このアナロジーをもう少し広げるデモを通じて、Kinesis Data Streams を使って、私の家庭で起こるような混乱など、多くのことが同時に起こるのに対応する方法を示したいと思います。
新しいプロダクトを構築していて、ローンチしたいと想像してみてください。本当に良いことの一つは、ローンチのソーシャルメディアセンチメントをフォローすることです。私たちはソーシャルメディアセンチメント分析ツールを用意しました。これはデモで、実際に自分で確認できるものです。GitHub にあり、デモへのリンクを少し後で提供します。
シナリオはこうです。午前11時にローンチして発表を行い、ある程度の利用が見られます。その後、5分過ぎたところで、セレブリティが私たちをピックアップして再投稿します。突然、このローンチの周りのアクティビティが急増し始めます。最終的に、ローンチから約45分後のバイラルモーメントに達し、ものすごい量のエンゲージメントが見られ、私たちのソーシャルメディア分析ストリーミングアプリケーションは本当にこれすべてに対応しようと必死です。そして最終的に、ハイプは落ち着き、私たちはトラフィックのベースラインに戻ります。これがアーキテクトが必要な理由です。
では、これをどのようにやるのかを見てみましょう。左側にはイベント プロデューサーがあります。これはソーシャル ネットワークからのデータ フィードになります。 このデータをフィードしているので、データのプロデューサーがこのデータを Kinesis Data Streams に送信しています。その後、ストリーム処理コンシューマーがあります。この場合、Lambda を使用して構築しました。これは大規模言語モデルを使用してセンチメント分析を行うなど、あらゆる種類のことを行うことができます。その後、データ結果をデータレイクに保存して、分析に使用したり、さらなるイベントをトリガーしたりします。Data Firehose を使用してそのデータを Amazon S3 テーブルに入れます。これは初期ローンチ後のフェーズ 1 です。
1 秒あたり約 100 メッセージが表示されており、これは処理しやすいです。特に悪いことはありません。この特定の負荷に対しては十分にスケーリングされています。 ここで ListShards API 呼び出しの結果を示します。4 つのシャードが開いているのが見えます。すべてが順調に進んでいます。その後、セレブリティが私たちを取り上げます。 彼らのフォロワーにこのローンチが通知されます。何が起こるかというと、エンゲージメントが非常に大幅に増加するのが見えます。
実際に起こっていることは、受信レコードがドロップされているということです。なぜなら、プロデューサーからの入力がストリームがスケーリングできる量より多いからです。ただし、オンデマンドを使用しているため、ストリームは自動的にスケーリングされます。最終的に、スロットリング レコードはベースラインに戻ります。元々持っていた 4 つのシャードが分割され、新しいシャードが作成されたのが見えます。これは素晴らしいことです。これはすべてスケーリングを自動的に処理しています。
バイラル モーメントに達すると、それらのシャードをさらに分割します。最終的にセンチメント レコードのピークが入ってきます。この特定のワークロードは現在、分析する必要があるソーシャル メディア投稿の 1 秒あたり約 50,000 レコードまでスケーリングされています。 これすべての終わりに、ストリームを 128 個のアクティブ シャードまで自動的にスケーリングしました。ただし、分割して閉じられたすべてのシャードが見えます。
合計すると、5,510 万レコードを処理しましたが、残念ながら 200 万レコードをドロップしてしまいました。失われたデータがかなりありました。残念ながら、この場合、初期スケールが十分ではなかったため、ソーシャル メディア センチメント分析アプリケーションはこのような結果になりました。
では、On-Demand Advantage とウォームスループットを使って同じことをやるとしたら、コンソールに入って、ストリームのウォームスループットについて、どのくらい欲しいかを指定することができます。これは単なる設定で、メガバイト毎秒で指定します。 コンソールのこのツールは本当に素晴らしくて、テストデータでも本番データでも使うことができます。ウォームスループットをどのくらいに設定すべきかについて、アイデアを得ることができます。
つまり、このくらいの負荷が予想されるけど、どのくらいに設定すればいいのか、という感じで言えるわけです。このプロセスを通じて手助けしてくれるので、かなり便利です。では、On-Demand Advantage でこれを始めると、128 個のシャードで始まるのが見えます。 最初は非常に少ない量のトラフィックしかないのに、セレブリティがこれを自分たちのネットワークで共有すると、トラフィックがスパイクするのが見えます。 ここの結果の終わりに到達すると、 5,720 万件のレコードを処理することができました。何も落とすことはありませんでした。ソーシャルメディア分析からはるかにクリーンなシグナルが得られました。新しいシャードを作成しなかったことが見えます。このような負荷を処理するために、完璧に適切にスケールアップされていました。
これは、ローンチのような予測できるビジネスイベントに対して、ウォームスループットがどのように役立つかについて、本当に興味深いデモンストレーションだったと思います。 前述したように、これは GitHub で利用可能です。興味があれば、QR コードのスナップを撮ることができます。チェックアウトして、自分で実行することができます。On-Demand Advantage がピークイベントをあなた自身にとって問題にならなくする方法の、本当に興味深い例です。
内部アーキテクチャの深層：shardの分割・統合による革新的なスケーリング設計
では、Ali に引き継いで、残りの under the hood の詳細について説明してもらいます。ありがとうございました。素晴らしい。John は Kinesis Data Streams についてたくさんの素晴らしい機能について言及しました。John が言及したように、私は AWS のストリーミングアーキテクトで、毎日顧客と一緒に仕事をしています。非常に多くの場合、自分たちで管理するストリーミングストレージをセットアップ、構成、微調整するのを手助けして、大きなイベントの時にスケールできるようにします。時々彼らはガイダンスが必要で、その時は彼らのインフラをスケールするのを手助けします。
非常に多くの場合、1 ギガバイト毎秒から 10 ギガバイト毎秒へのスケーリングには数週間の計画が必要です。それは私にとって、Kinesis ストリームがオペレーターが何もしなくてもそれをどのようにできるのかについて、好奇心の対象になりました。これは本当に興味深いです。ですから、エンジニアと会話をしました。連絡を取って、John の許可を得て、素晴らしい人たちと繋がって、このシステムをどのように設計したのか、そしてこれがどのように異なるのかについて、説明してもらえますかと聞きました。彼らは多くの under the hood の事実を共有してくれて、私は今日、初めてあなたたちにこれらすべての under the hood の詳細を共有するためにここにいます。
John の説明の復習ですが、Kinesis Data Streams の原則は、使いやすく、高い可用性を持ち、セキュアで、耐久性があり、非常に大規模な環境で低レイテンシーを実現することです。数百万のデータストリームと、また毎秒数億のリクエストに対応します。エンジニアたちが私に共有してくれたのは、彼らが最初に直面した難しい決断の一つが、システムがどのようにスケールすべきかということでした。John が述べたように、shard はスケールの単位であり、その後、水平スケーリングを行うわけです。簡単ですね？
さて、一つの実装方法として、他のストリーミングストレージでも同じように機能するものがありますが、単純に shard を追加するというアプローチがあります。つまり、shard 1 があって、その後、別の shard を追加するわけです。このアプローチにはいくつの問題があると思いますか？1 つ、2 つ、3 つ。私は 3 つ特定しました。1 つ目は、このスライドで示しているように、メッセージが順序外れになる可能性があります。なぜでしょうか？ちなみに、メッセージは同じキーを持っています。なぜでしょうか？
よく注意してみると、shard を導入した時に、producer がハッシュアルゴリズムを使ってキーを暗号化する部分が表示されていました。そして、ハッシュアルゴリズムがどのように範囲に分割するかに応じて、各範囲が各 shard にマップされ、メッセージはその shard に属し、その shard に送信されます。これがシステムがメッセージを分散する方法ですね。では、1 つの shard から 2 つに移行するとどうなるでしょうか？shard の範囲が分割され、1 つの shard に属していたすべてのメッセージが、別の shard の範囲に属する可能性があります。そうすると、同じキーを持つメッセージ C が別の shard に終わる可能性があります。consumer はそれらを順序外れで読み込むことになります。
2 つ目の問題は、スケールバックすることになっているのでしょうか？今、2 つの shard から 1 つに戻すことができますか？いいえ。そのデータはどうするのでしょうか？そのデータを何らかの方法で読み込むのでしょうか？
単純にそのデータを読み込んで、別の shard にマージすることはできません。別の問題があります。バックエンドノードの負荷を分散したい場合、これについては後で説明しますが、shard に非常に多くのデータが含まれることになり、shard は簡単に 10 テラバイト以上に達する可能性があります。この負荷を異なるワーカーノード間でどのようにバランスさせるのでしょうか？そのデータをあるノードから別のノードへ、そしてまた別のノードへコピーする必要があります。これには多くのリソースが必要で、多くの時間を消費し、予測可能であるという原則に違反します。
エンジニアたちは、shard がどのようにスケールするかについて、非常に革新的なアプローチを取りました。shard は単に別の shard を追加するのではなく、shard を分割します。つまり、アクティブな shard（親とも呼ばれます）がクローズされ、2 つの新しい shard（子とも呼ばれます）が追加されるということです。コンシューマーはまず親から messages を読み、その後、子から messages を読みます。
スクリーンショットを撮って stream の shard を説明すると、このような感じになります。親 shard がクローズされ、2 つの新しい shard が開かれています。同様に、スケールバックインしたい場合は、2 つの前の親 shard がクローズされ、その後、子の shard が追加されます。producer からのすべての messages は新しい shard（子の shard）に来て、その後、親 shard がクローズされます。コンシューマーは親 shard から messages を消費し終わるまで消費でき、その後、また分割したり、また統合したりできます。
このアーキテクチャの良い点は、messages が順序を失う問題を解決することです。このアプローチにより、データ損失なく、また message の順序に関する問題なく、スケールバックインできます。さらに、これらの shard のそれぞれは、異なるワーカーノード上にある可能性があります。分割または統合されない shard があり、そのサイズが何らかの制限に達している場合、それをクローズして、残りを別のワーカー上で新しい shard を開くために移動することができます。このようにして、バックエンドで均等な負荷分散を行っているため、データをあちこちにコピーする必要がありません。
高可用性とAPIレイヤー：3つのアベイラビリティーゾーンと分散キャッシュシステム
ここで何が起こるかを説明します。ワーカーは実際には、データを永続化するストレージノードです。ワーカーが 3 つのアベイラビリティーゾーンにあるため、それらに対して 3 つの列が表示されます。shard がクローズされると、新しい shard は別のワーカー上で開くことができ、他のワーカー上にレプリカがあり、その後、データはそこに行きます。前の shard はそこにありますが、クローズされています。producer はそれにデータを書き込むことができず、その後、新しいワーカーが書き込みを受け入れます。リーダーはまず親 shard を読み、その後、子の shard に来ます。
言及されたもう 1 つの原則は、Kinesis Data Streams は高可用性を備えていなければならないということです。高可用性とは、Kinesis Data Streams に書き込むデータのあらゆるコピーに対して、Kinesis Data Streams は実際に 3 つのコピーを保存し、それらを 3 つの異なるアベイラビリティーゾーンに配置するということです。なぜでしょうか？各アベイラビリティーゾーンは障害の封じ込めゾーンだからです。したがって、データは 3 つのアベイラビリティーゾーンで終わります。アベイラビリティーゾーン全体を失った場合でも、2 つのコピーが利用可能です。
先ほど説明したアーキテクチャと、ノードが3つのアベイラビリティゾーン全体にレプリケートされているという事実があります。これにより、いくつかの複雑性が生じます。これらのノードの1つは常にリーダーになります。つまり、書き込みを受け付けるノードであり、他の2つのノードはレプリカになります。コンシューマーとプロデューサーが読み書きをしたいときは、そのノードに接続する必要があります。接続する方法を見つける必要があります。 したがって、ディスカバリーを行う必要があります。どのように接続するか、DNSレコードは何か、IPアドレスは何か、そしてどのように接続するかを調べ出す必要があります。そして接続を確立します。レプリカまたはリーダーが失敗して、別のリーダーが引き継ぐ場合、それについても知る必要があり、その後新しい接続を確立できます。 これは Kinesis Data Streams が使いやすくあるべきという別の原則に違反しています。
したがって、エンジニアたちは前面のフリート全体を構築し、すべての複雑性を顧客から抽象化して、単一の API を公開することで対応しました。単一の API を通じて、様々なソースからデータを収集し、それをネットワークインフラストラクチャに取り込むことができます。ネットワークインフラストラクチャを管理する必要がなく、そのインフラストラクチャのスケールについて心配する必要がなく、API に対して発生する可能性のある DDoS 攻撃や悪意のあるアクションについて心配する必要もありません。その API は利用可能であり、複数のアベイラビリティゾーン全体でデータの自動バランシングとロードバランシングを処理します。
これで、顧客から来るリクエストがどのようにバックエンドノードにルーティングされるかを把握することが Kinesis Data Stream の責任になったため、 エンジニアたちは、あなたから隠されている3つの内部サービスを構築する必要がありました。キャッシュサービス、メンバーシップサービス、タイマーサービスです。これらについてはもっと詳しく説明します。顧客の責任もあります。コンシューマーとプロデューサーの責任は、シャードを認識することです。各 Kinesis Data Stream の下にシャードがあることを知ることです。そしてプロデューサーの場合は、シャードの制限を超えないようにデータを多くのシャード全体でロードバランシングすることです。コンシューマーの場合、責任は処理スレッドを認識することです。どのスレッドがどのシャードから読み込んでいるのか、そしてリース管理と呼ぶ何らかのプロセスを実行することですが、これについてはもっと詳しく説明します。
Kinesis Data Stream は、バックエンドノードのすべての側面を継続的に 監視します。別のコンシューマーを追加する場合、他のストリーミングストレージを使用する場合、容量計画をもう一度やり直す必要があります。なぜなら、入ってくるスループットを維持するのに十分なネットワークスループットとディスクスループットがあることを確認する必要があり、さらにすべてのバックグラウンド操作のためのバッファを持つ必要があり、さらに追加のコンシューマーが同じデータを読み込むための十分な容量を持つ必要があるからです。Kinesis DataStream では、そのような必要はありません。なぜなら、Kinesis DataStream には 最近立ち上がった enhanced fan-out という機能があり、単一の Kinesis Data Stream から最大50の異なるコンシューマーグループをサポートしているからです。
それでは、その API レイヤーについてもう少し詳しく説明しましょう。 Kinesis Data Stream を使用していると手を挙げた方々、ありがとうございます。しかし、今日あなたが書き込んでいるデータは、この部屋の他の誰かと同じノードに保存されている可能性が高いです。したがって、API は、入ってくるすべてのリクエストが認証されていることを確認し、アクセスしようとしているユーザーが指定されたリソースにアクセスする権限を持っていることを確認する必要があります。また、クォータを超えていないこと、バックエンドノードを過度に駆動していないことを確認する必要があり、データが転送中の暗号化または TLS で暗号化されていること、また顧客が管理するキーまたはバックエンドの AWS キーで暗号化されていることを確認する必要があります。
IPv6 を使用する要件がある場合、dual-stack API の一部として単一の API を使用すれば、AWS SDK を通じて IPv6 または IPv4 を取得でき、それを有効にすることができます。US government cloud で FIPS エンドポイントを使用する必要がある場合、FIPS エンドポイントで Kinesis data stream にアクセスできるので、それを自分で構築する必要はありません。アクセス制御を行いたい場合、ストリームレベルでリソースへのアクセスを制御するためのリソースベースのポリシーにアクセスでき、また IAM（AWS Identity and Access Management）を通じたアクセスポリシーにもアクセスできます。
フロントエンドサービスは、それぞれのフロントエンドサービスが毎秒数億回 これを行います。どのようにそれを行うのか見てみましょう。これらの API は、どのメンバーがどのリソースにアクセスしたいのか、そしてどのシャードがどのストリームに属し、どのメンバーに属しているのかに関する情報が必要です。その情報はまた、どのノードがリーダーノードで、どのノードがレプリカノードであるかも示しており、リクエストが来たときのルーティングはそのようにして行われます。 素朴なアプローチは、単にそれをデータベースに保存して、その後、入ってくるすべてのリクエストに対して、シャード ID または Kinesis stream ID で行って、データベースからその情報をロードして、リクエストをルーティングする必要があるノードを見つけて、それを行うというものです。
しかし、このデザインは機能しないことがわかっています。なぜでしょうか？大規模なデータストリームを構築している場合、サービスが信頼できることを確認する必要があるため、それよりもさらに大きなデータベースサービスが必要になるからです。また、データベースは通常二桁のレイテンシーを持つため、レイテンシーも非常に低くはありません。その後、API レイテンシーはそれよりもさらに高くなります。ですから、他のエンジニアが言うように、キャッシュをその前に置きましょう。
これは機能するでしょうか？まあ、これは確かにパフォーマンスを改善しましたが、常にそうとは限りません。キャッシュが失敗したらどうなるでしょうか？その場合、データストリームよりもさらに大きな容量を提供するキャッシュサービスが必要になり、また、キャッシュはさまざまな理由でデータベースと同期しなくなる可能性があります。そこでエンジニアたちは、データベースが実際にどのように機能するかの基本に立ち返りました。考えてみると、データベースが私たちにしてくれることは、すべての変更ログの一貫したビューまたはスナップショットを提供することです。入ってくるすべての変更ログに対して、読み取りが発生した時点で常に最新で正確なスナップショットを提供します。
そこでエンジニアたちは、それをすべての異なるコンポーネント、フロントエンドホストとキャッシュレイヤーにも複製したらどうだろうかと考えました。その場合、これらの各コンポーネントは独立してそのスナップショットを構築できるので、リクエストが来たときに、ワーカーのメモリ内のキャッシュからその情報を読み取ることができます。このデザインは機能するでしょうか？完全にはそうではありません。近づいていますが、完全にはそうではありません。
このデザインの問題は、分散システムを運用しているということです。そして、この分散システムは様々な理由で同期が取れなくなる可能性があります。ノードが失敗することもあれば、チェンジログが複製されて2回処理されることもあります。処理順序が狂うこともあります。そしてシステムが同期を失う可能性があります。そのため、エンジニアたちはシャードマップエンティティに追加情報を加えました。このレコードがどのくらいの期間有効なのかを示す情報です。ノードがレコードを持っている場合、この時間が経過したかどうかもチェックします。この時間が経過していれば、そのレコードは古いということなので、新しいレコードを読み込みに行く必要があります。
では、このデザインは今うまく機能するでしょうか？いいえ、まだ1つ解決すべき問題があります。分散システムでは時間の流れが異なるということです。時間を比較したいときは、信頼できるクロックと比較する必要があります。信頼できるクロックとは何でしょうか？異なるコンポーネントのシステム時刻は十分に信頼できるでしょうか？いいえ、ここではミリ秒が重要です。毎秒数億のリクエストを処理しています。間違ったノードにリクエストが送られてエラーが発生し、その後再試行する必要があるとしたら、それは信頼できるシステムではありません。
そのため、エンジニアたちは Timer Service という別のサービスを構築しました。Timer Service は、これらすべての異なるコンポーネント間における時間の信頼できる情報源です。システムに表示されているシステム時刻が何であろうと関係ありません。その時間を Timer Service の時間と比較する必要があります。彼らは常に Timer Service をポーリングしてそれを更新し、その後それを time to live と比較するために使用します。そのレコードが古い場合は、キャッシュから取得します。キャッシュにない場合は、データベースから取得します。では、このデザインは機能するでしょうか？はい、これは機能します。
コンシューマーの責任とKinesis Client Library：リース管理と負荷分散の仕組み
では、コンシューマーの責任について話しましょう。先ほど述べたように、コンシューマーはシャードを認識する必要があります。なぜなら、データを分散的に処理しているからです。コンシューマーが管理する必要がある特定の側面があります。1つは、どの処理スレッド、どのワーカーがどのシャードから消費しているかという状態です。シャードとコンシューマーのインスタンスまたはコンシューマー処理スレッド間のマッピングをどこかに保持する必要があり、その後それを常に監視・管理する必要があります。そのため、コンシューマーにはいくつかのベストプラクティスがあります。ゼロからコンシューマーアプリケーションを自分で書いていて、独自のクライアントライブラリを書きたい場合は、適切なハンドリングを行い、クライアントを監視して、状態の追跡をしっかり行うようにしてください。
これらのネイティブサービスまたはサードパーティアプリケーションを使用している場合は、Kinesis Data Streams とこれらのアプリケーション間に既に統合があります。このネイティブ統合があるため、コードを書く必要はありません。データが Kinesis Data Streams に到着するとすぐに、これらのサービスまたはサードパーティを使用してそれを消費できます。
AWS Lambda を使いたくない場合は、Kinesis client library がすべての重い処理と状態管理を引き受けてくれます。本当に自分でやりたいのでなければ、Kinesis client library はあなたのコードにパッケージ化して、Kinesis Data Streams から消費するカスタムコードを書くための最も簡単な方法です。
最近、Kinesis チームは Kinesis client library のバージョン 3 をリリースしました。本当にたくさんの改善が入っています。その改善の中でも、私のお気に入りは、処理スレッドのどれが過負荷になっているかを検出して、さらに負荷を分散させるというものです。基本的には、あるワーカーからリースを取得して、別のワーカーに与えるんです。ワーカー間で負荷を分散させるので、コスト削減のメリットが得られます。
Kinesis client library を使う場合、状態は 3 つの別々の DynamoDB テーブルで管理されます。状態テーブルはどのワーカーがリーダーであるかを追跡します。リーステーブルはどのワーカーと処理スレッドがどのシャードから消費しているのか、そしてそのサブスクリプションとマッピングがどのように見えるかを追跡します。そして、Kinesis client library バージョン 3 が使用するメトリクステーブルがあり、これはどのワーカーが過負荷になっているかを知るためのもので、さらに負荷を分散させます。また、すべてのクライアント関連のメトリクスを Amazon CloudWatch に出力するので、その統合について心配する必要もありません。
左側はポーリングを担当するワーカーです。右側は処理スレッドで、これらのスレッドは互いに分離されています。マルチスレッドプログラミングをやったことがある人なら、メモリリークやその他の問題なしにマルチスレッドプログラミングを実行することがいかに難しいかを知っています。また、タイマーのように機能し、すべてを実行するスケジューラもあります。
リース管理レコードがどのように見えるかをここに示します。キーとチェックポイントがあります。各処理スレッドは進行するにつれて、チェックポイントを頻繁に更新し、チェックポイント情報はリース管理テーブルに保存されます。また、タイムツーリブのような、リース有効期限時間もあります。それが果たす目的は非常に重要です。リースレコードが更新されていない場合、つまりこのタイムスタンプが最新でない場合、それは処理スレッドがサイレントに失敗したことを意味します。システムがそれを検出して、別のワーカーが引き継ぎます。誰が勝つか、明らかに最初にリースを取得した者が、別のワーカーから引き継ぎます。
リーダーが失敗した場合、別のワーカーが基本的に引き継ぎます。ワーカーが失敗した場合、リーダーは前のワーカーからリースを取得して、別のワーカーに与えることができます。これが分散システムとステートフルな分散コンピュートシステムがどのように機能すべきかということであり、Kinesis クライアントライブラリでもそのように機能します。
プロデューサーのベストプラクティス：バッチ処理、集約、圧縮によるパフォーマンス最適化
では、プロデューサーについて話しましょう。プロデューサーもシャードを異なる方法で認識する必要があります。プロデューサーが1つのレコードごとに API コールを行うのは効率的でしょうか？多くのお客様と話をすると、それは効率的ではないことを認めていますが、それでもそうしている理由は、そうすることでレイテンシーが低くなると考えているからです。実は、ネットワークコールは安くもなく、無料でもありません。ネットワークコールにはオーバーヘッドがあります。すべてのリクエストは認証される必要があり、アクセス制御がチェックされ、アクセス制限に対してチェックされ、その後ルーティングされます。1つのレコードごとに1つのリクエストを送信している場合、ちなみにそのレコードは非常に小さいのですが、それは効率的に使用していません。
お客様には、レコードをより大きなペイロードにバッチ処理し、可能であれば非同期操作を実装することをお勧めします。
Kinesis Data Streams には、レコードを書き込むための2つの API があります。PutRecord と PutRecords です。可能な限り PutRecords API を使用して、複数のレコードを1つのネットワークコールで送信することをお勧めします。PutRecords では、最大500個のレコードを書き込むことができます。そのレコードのサイズはどのくらいであるべきでしょうか？それについてはもっと詳しく説明します。
ネイティブサービスまたはネイティブ統合を備えたサードパーティサービスを使用している場合、Kinesis Data Streams にデータを書き込むのに役立つネイティブコネクタがあります。これらのベストプラクティスはすでに組み込まれています。しかし、自分でコードを書いている場合、API を呼び出したい場合は、非常に小さいレコードがある場合は、集約も実装するようにしてください。レコードサイズは約10キロバイト程度であることをお勧めします。レコードサイズがそれより大幅に小さい場合、1つのテクニックは、より小さいレコードから大きなペイロードを構築することです。
では、クリックストリームレコードが400バイトだけだとしましょう。これを10 キロバイトのバッチに集約することができます。さらに圧縮することもできるので、より多くのクリックストリームレコードを詰め込んで、約10キロバイトまで圧縮することができます。その後 PutRecords APIを使用して、すべてを1つのネットワークコールで送信することができます。直感的ではありませんが、この方法を使うと実はレイテンシーとスループットの両方が向上します。なぜなら、小さなクリックストリームデータの非常に大きなスループットがある場合、ネットワークオーバーヘッドがパイプライン全体のスループットの移動のボトルネックになるからです。
これらのベストプラクティスについて心配したくなく、常に従われ、正しく実装されることを確保したい場合は、Kinesis Producer Library を使用することができます。これはもう1つのライブラリで、オープンソースです。すぐにリンクを提供します。これを使用して、小さなレコードをより大きなバッチに集約し、さらに圧縮することができます。有効または無効にできるオプションがあります。より大きなバッチに入れたら、バッファがあってそれらをバッファリングし、そのバッファは信頼性があります。ノードが失敗した場合、システムはレコードがそこにあることを保証します。
collector と retrier と limiter は、スロットリングに達したときに役立ちます。バッファが Kinesis Data Stream にフラッシュされるようにリトライします。limiter は、Kinesis Data Stream が課す割り当てと制限を超えていないことを確保します。スロットリングされたレコード数、通過したレコード数、成功したレコード数、およびパイプラインのレイテンシーに関するメトリクスを Amazon CloudWatch に出力します。プロデューサーの多くのインスタンスにわたって確実に動作することができます。
まとめ：デモの実践とリソース紹介、フィードバックのお願い
追加のリソースを ここに残しておきます。Kinesis Data Streams についてもっと詳しく知りたい場合、詳細を知りたい場合は、最初のQRコードにアクセスできます。2番目のものは、今日共有したデモで、GitHub へのリンクに移動してデモを試すことができます。Kinesis Producer Library または Kinesis Client Library のコードを読むか、プロジェクトを見つけることについてもっと詳しく知りたい場合は、右側の2つのQRコードにアクセスして、GitHub に移動して、それらをチェックアウトすることができます。
このセッション後にしてほしいことは、デモを試して、Kinesis Data Streams が10,000から100,000、または500,000、さらには数百万レコード/秒のピークにどのように対応できるかを確認することです。 今日共有したベストプラクティスの多くが含まれている、あなたの環境でのライブデモでパイプライン全体を監視してください。同様のアプリケーションを書いている場合、またはそれをプロジェクトのベースラインとして使用して、自動的にスケールアウトするサーバーレスパイプラインを構築したい場合は、これらのコンポーネントの多くを再利用できます。
もう一つのお願いとしては、皆さんに送られてくるリンクからアンケートに回答していただきたいということです。というのも、そのアンケートは私たちにとって本当に重要なものなのです。皆さんからのフィードバックを使って、コンテンツを改善し、今後のセッションをより良いものにしていくためです。そのアンケートは常に私たちにとって重要です。それでは、本日お越しいただき、夜遅くまで一緒にいてくださり、このプレゼンテーションを最後まで聞いていただいた皆さんに、心からお礼を申し上げます。ありがとうございました。
※ こちらの記事は Amazon Bedrock を利用し、元動画の情報をできる限り維持しつつ自動で作成しています。
Discussion