エラーメッセージ
AWS Kinesisにデータをプッシュしていると、突然ログが次のようなエラーで埋め尽くされることがあります。
ProvisionedThroughputExceededException: Rate exceeded for shard shardId-000000000000 in stream <stream-name> under account <account-id>.
これは単なる警告ではありません。パフォーマンスの限界に達したため、Kinesisがデータを破棄していることを意味します。
何が起きているのか？
ストリーム内の各シャードを、固定のサイズを持つ専用のパイプだと考えてください。AWSはこれらのパイプに対して、2つの厳格な速度制限を設けています。
- 取り込み帯域幅： 1秒あたり1 MB
- 書き込み頻度： 1秒あたり1,000レコード
いずれかを超えると、AWSはリクエストをスロットリング（制限）します。これは通常、トラフィックの総量がシャード数を超えたか、「ホットシャード（hot shard）」が発生している場合に起こります。ホットシャードは、パーティションキーが適切に分散されておらず、他のシャードがアイドル状態であるにもかかわらず、特定の1つのシャードだけに負荷が集中することで発生します。
ステップ1：ホットシャードを特定する
スケールアウトにはコストがかかるため、まずはデータの分散状況を確認しましょう。例えば、Region_IDをパーティションキーとして使用しており、US-EAST-1リージョンの負荷が他の10倍高い場合、ストリームの他の部分に空きがあっても、その特定のシャードがパンクしてしまいます。
CloudWatchコンソールを開き、以下の特定のメトリクスを詳しく確認してください。
IncomingBytesおよびIncomingRecords（ShardIdでフィルタリング）。WriteProvisionedThroughputExceeded。
外れ値を探します。もし1つのシャードが1,000レコード/秒に達している一方で、他のシャードが50レコード/秒であれば、それはキャパシティの問題ではなく、パーティションキーの問題です。
ステップ2：パーティションキーを再設計する
ここでは「高カーディナリティ（high cardinality）」が重要です。利用可能なすべてのシャードにデータを均等に分散させるために、数千ものユニークな値を持つキーを使用する必要があります。Store_Locationのような大まかなカテゴリではなく、UUIDや、User_ID + Timestampのような複合キーを検討してください。
キーがどのように分散されるかを確認する必要がある場合、ハッシュ化が確実な近道です。例えば、データのSHA-256ハッシュを生成すると、ほぼ完璧なランダム分散が保証されます。ToolCraftのHash Generatorを使用して、さまざまな入力からバランスの取れたシャードに必要なユニークな文字列がどのように生成されるかをテストできます。
ステップ3：エクスポネンシャルバックオフとリトライを追加する
トラフィックが常に一定であることは稀で、バースト的に発生します。アプリケーションは、一時的な制限に直面したときに一時停止して再試行できるレジリエンス（回復力）を備えている必要があります。AWS SDKはある程度のリトライを処理しますが、カスタムループを記述することで、「ジッター（jitter）」や遅延をより細かく制御できるようになります。
以下は、基本的なバックオフ戦略を実装したboto3を使用するPythonの例です。
import boto3
import time
import random
from botocore.exceptions import ClientError
kinesis = boto3.client('kinesis', region_name='us-east-1')
def put_record_with_retry(data, partition_key):
attempt = 0
max_attempts = 5
while attempt < max_attempts:
try:
return kinesis.put_record(
StreamName='my-data-stream',
Data=data,
PartitionKey=partition_key
)
except ClientError as e:
if e.response['Error']['Code'] == 'ProvisionedThroughputExceededException':
attempt += 1
# 少量のランダムなジッターを加えたエクスポネンシャルバックオフ
delay = (2 ** attempt) + random.uniform(0, 1)
print(f"Throttled. Retry {attempt}/{max_attempts} in {delay:.2f}s...")
time.sleep(delay)
else:
raise e
raise Exception("Failed after max retries")
ステップ4：ストリームをスケールする（リシャーディング）
単純に、より大きなパイプが必要な場合もあります。CloudWatchで、すべてのシャードが常に800〜900レコード/秒付近で推移していることが示されている場合は、リシャーディング（resharding）のタイミングです。このプロセスにより、ストリームをオフラインにすることなくキャパシティを追加できます。
AWS CLIを使用して、シャード数を即座に増やします。
aws kinesis update-shard-count \
--stream-name my-data-stream \
--target-shard-count 10 \
--scaling-type UNIFORM_SCALING
シャード数を2倍にすると、1時間あたりのコストも2倍になることに注意してください。過剰なプロビジョニングにならないよう、使用状況を監視してください。
修正を確認する方法
コードをデプロイして終わりではありません。エラーが解消されたことを確認するために、以下の3つの領域を監視してください。
- CloudWatchの傾向:
WriteProvisionedThroughputExceededのカウントがほぼゼロに低下していること。 - コンシューマーの状態: LambdaやKCLのラグを確認してください。プロデューサーが頻繁にリトライしている場合、コンシューマーが処理するデータが不足する可能性があります。
- エラーログ: ログに対してgrepやInsightクエリを実行します。
ProvisionedThroughputExceededExceptionは、予期しない大規模なスパイク時にのみ発生する状態であるべきです。
エキスパートからのヒント
- 書き込みをバッチ化する:
put_recordの使用をやめ、put_records（複数形）を使用して1回のリクエストで最大500レコードを送信します。これにより、HTTPオーバーヘッドを大幅に削減できます。 - Kinesis Producer Library (KPL) を試す: Javaユーザーにとって、KPLは非常に強力です。バックグラウンドで複雑なバッチ処理やリトライを処理し、プロデューサーの効率を大幅に向上させます。
- オンデマンドモードを検討する: トラフィックの変動が激しく、手動でのスケーリングを避けたい場合は、**オンデマンド（On-Demand）**モードに切り替えてください。GBあたりの単価は高くなりますが、AWSがシャードの管理を自動的に行います。