Amazon Kinesisに接続する
Structured Streaming を使用して、Amazon Kinesis との間でデータの読み書きを行います。
Databricksでは、すべてのS3トラフィックがAWSネットワーク上でルーティングされるように、S3 VPC Endpointを有効にすることをお勧めします。
ストリームを中断または再起動することなくシャード数を増やすことで、Structured Streaming を使用してリシャードできます。
クエリーのレイテンシのトラブルシューティングに関する推奨事項については、Kinesisのレイテンシを短縮するための推奨事項を参照してください。
認証
Kinesisは、Unity Catalog接続、サービス資格情報、またはインスタンスプロファイルやアクセスキーなどの代替メソッドによる認証をサポートしています。認証を参照してください。
スキーマ
Kinesis は次のスキーマでレコードを返します：
列 | Type | 説明 |
|---|---|---|
| string | レコードが割り当てられているシャードを識別するパーティションキー。 |
| binary | 不透明なバイナリとしての、レコードのデータ BLOB。 |
| string | レコードの読み取り元となった Kinesis ストリームの名前または ARN。 |
| string | レコードの読み取り元となったシャードの ID。 |
| string | シャード内におけるレコードの一意の識別子。 |
| timestamp | レコードがストリームに挿入された概算時間。 |
data 列のデータを逆シリアル化するには、フィールドを文字列にキャストします。
クイックスタート
次のノートブックは、Kinesisを使用した構造化ストリーミングを使用して WordCount を実行する方法を示しています。
Kinesis WordCount with Structured Streaming ノートブック
Kinesis オプションの設定
Databricks Runtime 13.3 LTS以降では、KinesisでTrigger.AvailableNowを使用できます。Kinesis レコードを増分バッチとして取り込むを参照してください。
Databricks Runtime 16.1以降では、streamARNを使用してKinesisソースを識別できます。すべてのDatabricks Runtimeバージョンにおいて、streamNameまたはstreamARNのいずれかを指定する必要があります（両方を指定することはできません）。
アクティブなストリーミング クエリーに対して、streamNameとstreamARNを切り替えないでください。Databricksでは、ストリームの途中でこれらのオプションを切り替えることはサポートされていません。クエリーを再起動すると、レコードの重複やデータ損失が発生する可能性があります。streamName から streamARN に切り替えるには、新しいチェックポイントディレクトリを使用して新しいストリーミングクエリーを開始してください。
オプションの完全なリストについては、Kinesisを参照してください。
ストリームソースの追加または削除
Databricks Runtime 19以降では、Sparkオプションの streamName または streamARN を使用して、Structured Streamingクエリーの Kinesis ソースストリームを変更できます。
ストリームの追加
ストリームを追加するには、それを streamName または streamARN オプションリストに含め、ストリームを再起動します。新しいストリームソースごとに、クエリーはソースのシャードから利用可能な最も古いオフセットを読み取ります。
次の例では、 streamName オプションを使用して、以前に stream1 および stream2 を読み取っていたクエリーに stream3 Kinesis ソースを追加します。
df = (spark.readStream
.format("kinesis")
.option("streamName", "stream1,stream2")
.load()
)
df.stop()
df = (spark.readStream
.format("kinesis")
.option("streamName", "stream1,stream2,stream3") # Previous value was "stream1,stream2"
.load()
)
ストリームの削除
By default、 streamName または streamARN オプションリストから Kinesis ストリームソースを削除すると、再起動時にクエリーは KINESIS_SOURCE_STREAMS_REMOVED_ON_RESTART エラーで失敗します。これにより、クエリーが削除されたストリームからの未読レコードを警告なしにスキップしないことが保証されます。
Kinesis ストリームソースを削除するには、次の操作を行います。
-
クエリーを停止します。
streamNameまたはstreamARNからストリームを削除し、failOnDataLossをfalseに設定します。たとえば、streamNameをstream1,stream2,stream3からstream1,stream2に変更します：Pythondf = (spark.readStream
.format("kinesis")
.option("streamName", "stream1,stream2")
.option("failOnDataLoss", "false")
.load()
) -
クエリーを再起動します。更新されたチェックポイントには、
stream3のオフセットが含まれなくなりました。このクエリーは、stream3からレコードを読み取らなくなりました。 -
Set
failOnDataLossback totrueand restart using the same checkpoint.This restores data-loss checks.
For more information about how failOnDataLoss affects クエリー behavior, see Handle data loss.
By default, Kinesisストリームソースを削除しても、そのストリームソースのEnhanced Fan-Out (EFO) コンシューマーは登録解除されません。そのため、クラウドプロバイダーから引き続きコストが発生する可能性があります。クエリーの停止時にコンシューマーを登録解除するには、requireConsumerDeregistrationオプションをtrueに設定します。See Kinesis.
コンシューマーを直接管理する方法については、ストリーミングクエリー読み取り用にKinesis Enhanced Fan-Out (EFO) を設定するを参照してください。
低遅延のモニタリングとアラート
アラートのユースケースでは、低レイテンシーが求められます。レイテンシーを最小限に抑えるには：
- ストリーミングクエリがKinesisストリームの唯一のコンシューマーであることを確認し、フェッチ性能を最適化し、Kinesisのレート制限を回避してください。
- オプション
maxFetchDurationを 200ms などの小さな値に設定して、フェッチされたデータを可能な限り迅速に処理します。このオプションはトレードオフの関係にあります。各バッチで最新のレコードが消費されることを保証するよりも、バッチごとの処理速度を優先します。例えば、Trigger.AvailableNowを使用する場合、値が小さいと、クエリーが Kinesis ストリーム内の最新のレコードから遅延する可能性があります。 - オプション
minFetchPeriodを210msに設定すると、可能な限り頻繁にデータを取得します。 - Databricks Runtime 19 以降では、
maxPartitionsを設定して、Kinesis に割り当てられる Spark タスクの数を制御します。Spark はmin(active shards, maxPartitions)タスクを使用し、各タスクが 1 つ以上のアクティブなシャードを処理します。maxPartitionsとshardsPerTaskの両方を設定することはできません。 - オプション
shardsPerTaskを設定するか、# cores in cluster >= 2 * (# Kinesis shards) / shardsPerTaskとなるようにクラスターを設定します。これにより、バックグラウンドのプリフェッチタスクとストリーミングクエリータスクが同時に実行されることが保証されます。
Kinesis メトリクスの監視
Kinesisは、各ワークスペースごとにストリーム開始時に消費者が遅延するミリ秒数を報告します。avgMsBehindLatest、maxMsBehindLatest、minMsBehindLatestのメトリクスは、ストリーミングクエリープロセスのすべてのワークスペースにおける平均、最小、最大ミリ秒を提供します。Databricksの「Structured Streaming クエリーのモニタリング」を参照してください。
ソースメトリクスを表示
次のいずれかの方法を使用してソースメトリクスにアクセスします：
- Raw Data
- Python
ノートブックで Stream を実行している場合、ストリーミングクエリの進行状況ダッシュボードの 生データ tab でメトリクスを確認します。次の例は、ソースメトリクスを示しています：
{
"sources": [
{
"description": "KinesisV2[stream]",
"metrics": {
"avgMsBehindLatest": "32000.0",
"maxMsBehindLatest": "32000",
"minMsBehindLatest": "32000"
}
}
]
}
After a Trigger completes, access the source metrics through lastProgress:
progress = query.lastProgress
print(progress["sources"][0]["metrics"])
Amazon CloudWatch で監視する
次の Amazon CloudWatch メトリクスを監視することもできます：
メトリクス | モード | 説明 |
|---|---|---|
| ポーリング |
|
| ポーリング |
|
| EFO | シャードから受信したバイト数。 |
| EFO | シャードから受信したレコード。 |
| ポーリング | 返された最後のレコードの経過時間。値が増加している場合は、コンシューマーが遅れていることを示しています。 |
| ポーリング | シャードの読み取り容量を超過したためにスロットリングされた読み取りの数。 |
低throughputの診断
throughput低下の原因を診断するには、Kinesis ソースメトリクスで maxMsBehindLatest を監視し、リアルタイムモードでのポーリングクエリーについては numAwsRateLimitErrors を監視します。また、タスク使用状況メトリクスの busyTimeFraction も監視します。適切なアクションを特定するには、次の表を使用します。
メトリクスのステータス | 意味 | 解決方法 |
|---|---|---|
| クエリーは最新の状態です。 | 操作は必要ありません。 |
リアルタイムモードでのポーリングクエリーの場合、 | AWSのスロットリングにより、ポーリングの throughput が制限されます。 | Kinesis シャードを追加するか、 |
| Spark タスクが十分に活用されています。 | コンピュートとタスクの数を増やします。 |
レート制限エラーが多すぎることによる速度低下を回避
コネクタは、レート制限エラーが発生するたびにKinesisから読み取るデータ量を半分に減らし、このイベントをログにメッセージとして記録します： "Hit rate limit. Sleeping for 5 seconds."
ストリームが追いついている間にこれらのエラーが見られることがあります。ストリームが追いついた後にこれらのエラーが表示される場合は、AWS での Kinesis 容量を増やすか、Spark のプリフェッチオプションを調整してワークロードを調整する必要があるかもしれません。
各 Kinesis シャードは、ポーリングリーダー全体で 1 秒あたり 5 回の GetRecords 呼び出しをサポートします。(N) ポーリングクエリーが同じストリームを読み取る場合は、各シャードへの呼び出しがこの制限内に収まるように、minFetchPeriod を少なくとも 200ms * N に設定します。
.option("minFetchPeriod", "400ms") # Two polling queries read the same stream.
Kinesis レコードを増分バッチとして取り込む
Databricks Runtime 13.3 LTS 以降、Databricks は、増分バッチセマンティクスのために、Kinesis データソースでの Trigger.AvailableNow の使用をサポートしています。以下は、基本的な設定に関する説明です：
- マイクロバッチの読み取りが available now モードでトリガーされると、Databricks クライアントによって現在の時刻が記録されます。
- Databricks は、この記録時刻と前のチェックポイントの間のタイムスタンプを持つすべてのレコードについて、ソースシステムにポーリングします。
- Databricks は、
Trigger.AvailableNowセマンティクスを使用してこれらのレコードを読み込みます。
Databricksは、ストリーミングクエリーの実行時にKinesisストリーム内に存在するすべてのレコードを消費するために、ベストエフォート型のメカニズムを使用します。Timestampにわずかに差異が生じていたり、データソース内の順序が保証されていなかったりするため、Triggerされたバッチに一部のレコードが含まれない可能性があります。含まれなかったレコードは、次にTriggerされるマイクロバッチで処理されます。
レコードが存在するにもかかわらず、クエリーが Kinesis ストリームからのレコード取得に失敗し続ける場合は、maxFetchDuration の値を増やしてみてください。
AvailableNow：増分バッチ処理を参照してください。
データ損失の処理
Databricks Runtime 19 以降で使用できます。データ損失が検出されたときにストリーミング クエリーでエラーが発生するかどうかを設定するには、failOnDataLoss オプションを使用します。Kinesis を参照してください。
This option applies only to the specific クエリー.Other Kinesis クエリー use their own option value or, if omitted, the Spark configuration.
ワークロードがレコードの欠落を許容できる場合にのみ、failOnDataLoss を使用してください。これを誤って使用すると、データが完全に失われる可能性があります。レコードの欠落を許容できない場合は、新しいチェックポイントでストリームを再起動し、すべてのレコードを再処理してください。
Databricksでは、データ損失問題に対する一時的な緩和策としてのみこれを使用することを推奨しています。Kinesisの保持期間が短すぎるなど、根本原因を調査して修正します。
ストリーミングクエリーが読み取る前に Kinesis シャードのレコードが期限切れになった場合、または同じ名前で Kinesis ストリームを削除して再作成した場合、クエリーは KINESIS_COULD_NOT_READ_SHARD_UNTIL_END_OFFSET エラーで失敗します。KINESIS_COULD_NOT_READ_SHARD_UNTIL_END_OFFSET を参照してください。
By default, ストリーミング クエリー fail when they detect potential data loss.To configure the クエリー to skip unreadable records and continue processing, set the failOnDataLoss option to false on the Kinesis read and restart the クエリー.
failOnDataLoss Spark オプションは、spark.databricks.kinesis.failOnDataLoss Spark 構成よりも優先されます。代わりに Spark 構成を使用するには、failOnDataLoss オプションを省略し、クラスターの Spark 構成で spark.databricks.kinesis.failOnDataLoss を false に設定して、クラスターを再起動します。
Kinesis への書き込み
次のコードスニペットを、Kinesisにデータを書き込むためのForeachSinkとして使用します。これには Dataset[(String, Array[Byte])] が必要です。
以下のコードスニペットは、「厳密に 1 回」ではなく、「 最低 1 回 」のセマンティクスを提供しています。
Kinesis Foreach シンクノートブック
Kinesis のレイテンシーを削減するための推奨事項
このセクションでは、Kinesisストリームのさまざまなレイテンシーの原因をトラブルシューティングするための推奨事項を説明します。
Kinesisソースは、バックグラウンドスレッドでSparkジョブを実行し、Kinesisデータを定期的にプリフェッチして、Sparkエグゼキューターのメモリにキャッシュします。各プリフェッチステップが完了すると、ストリーミングクエリーはキャッシュされたデータを処理できるようになります。プリフェッチステップは、観測されるエンドツーエンドのレイテンシーとthroughputに大きな影響を与えます。
プリフェッチのレイテンシーを短縮
クエリーのレイテンシーを最小限に抑え、リソースの使用量を最大化するように最適化するには、次の計算式を使用します。
total number of CPU cores in the cluster (across all executors) >= total number of Kinesis shards / shardsPerTask。
minFetchPeriod ReadProvisionedThroughputExceeded に達するまで、Kinesis シャードに対して複数の GetRecords API 呼び出しを作成できます。例外が発生した場合でも、コネクタが Kinesis シャードの利用率を最大化するため、問題ではない可能性があります。
ディスクへのspillを回避
Kinesis ストリームのデータ量が急増した場合、割り当てられたバッファ容量が一杯になり、新しいデータを追加するのに十分な速さで空き容量が確保されない可能性があります。Spark はバッファからディスクへデータをspillするため、ストリーム処理が遅延し、Logsに次のようなメッセージのイベントが表示されます：
./log4j.txt:879546:20/03/02 17:15:04 INFO BlockManagerInfo: Updated kinesis_49290928_1_ef24cc00-abda-4acd-bb73-cb135aed175c on disk on 10.0.208.13:43458 (current size: 88.4 MB, original size: 0.0 B)
spillを回避するには、ノードを追加するかノードあたりのメモリを増やすことでクラスターのメモリ容量を増やすか、構成パラメーター fetchBufferSize を減らしてください。
中断された S3 書き込みタスク
ストリーム処理の進行を妨げる中断されたタスクを終了させるには、Sparkの投機的実行（speculation）を有効にします。タスクが過度に終了されないように、この設定の分位点と乗数を慎重に調整してください。Databricksでは、spark.speculation.multiplierを3に、spark.speculation.quantileを0.95に設定し、必要に応じて調整することを推奨しています。
ステートフルストリームでのチェックポイントによる遅延の削減
Databricks では、ステートフルなストリーミングクエリに対して、RocksDB と変更ログのチェックポイント設定を併用することをお勧めします。変更ログのチェックポイント設定を有効にするを参照してください。
ストリーミングクエリー読み取り用にKinesis Enhanced Fan-Out (EFO) を設定する
Databricks Runtime 11.3以上では、Databricks Runtime KinesisコネクタがAmazon Kinesisの拡張ファンアウト（EFO）機能の使用をサポートします。
Kinesis enhanced fan-outは、シャードあたり、コンシューマーあたり2MB/秒の専用throughput（ストリームあたり最大20コンシューマー）を提供し、プルモードではなくプッシュモードでレコードを配信します。
デフォルトでは、EFOモードで構成されたStructured Streamingクエリーは、専用のthroughput、一意のコンシューマー名、およびコンシューマーARN（Amazon Resource Name）を持つコンシューマーとしてKinesis Data Streamsに自身を登録します。
By default, Databricks はストリーミングクエリー ID に databricks_ プレフィックスを付けて、新しいコンシューマーに名前を付けます。必要に応じて、consumerNamePrefix または consumerName オプションを指定して、この動作をオーバーライドできます。consumerName は、文字、数字、および _ . - の特殊文字のみを含む文字列である必要があります。
クエリーの再起動時に、Kinesis ソースはポーリングモードを使用して、コミットされていない最新のバッチが存在する場合はそれをリプレイします。ストリームがコミットされていないバッチをリプレイした後、ソースは後続の読み取りのためにEFOモードに切り替わります。
登録されたEFOコンシューマーには、Amazon Kinesisでの追加料金が発生します。クエリーのティアダウン時にコンシューマーを自動的に登録解除するには、requireConsumerDeregistrationオプションをtrueに設定します。Databricksは、ドライバーのクラッシュやノードの障害などのイベントによる登録解除を保証できません。ジョブが失敗した場合、Kinesisの過剰請求を防ぐために、登録されたコンシューマーを直接管理することをDatabricksは推奨します。
Databricks ノートブックを使用したオフラインコンシューマー管理
AWSアカウントコンソールで手動でコンシューマーを構成する代わりに、AWSKinesisConsumerManagerユーティリティを使用して、Kinesisデータストリームのコンシューマーをプログラムで登録、一覧表示、または登録解除します。たとえば、このユーティリティを使用して新しいストリームのコンシューマーを作成します。ストリームを完全に停止する予定がある場合は、このユーティリティを使用してAWS内のコンシューマーを削除します。
コンシューマーマネージャーユーティリティは、コンピュートが専用アクセスモードに設定されている Scala でのみ使用できます。アクセスモードを参照してください。
Databricksノートブックでこのユーティリティを使用するには：
-
アクティブなクラスターに接続された新しいDatabricksノートブックで、必要な認証情報を提供して
AWSKinesisConsumerManagerを作成します。Scalaimport com.databricks.sql.kinesis.AWSKinesisConsumerManager
val manager = AWSKinesisConsumerManager.newManager()
.option("serviceCredential", serviceCredentialName)
.option("region", kinesisRegion)
.create() -
コンシューマーをリストして表示します。
Scalaval consumers = manager.listConsumers("<stream name>")
display(consumers) -
特定のストリームのコンシューマーを登録します。
Scalaval consumerARN = manager.registerConsumer("<stream name>", "<consumer name>") -
特定のストリームのコンシューマーの登録を解除します。
Scalamanager.deregisterConsumer("<stream name>", "<consumer name>")