Amazon Kinesisに接続する

Structured Streaming を使用して、Amazon Kinesis との間でデータの読み書きを行います。

Databricksでは、すべてのS3トラフィックがAWSネットワーク上でルーティングされるように、S3 VPC Endpointを有効にすることをお勧めします。

注記

ストリームを中断または再起動することなくシャード数を増やすことで、Structured Streaming を使用してリシャードできます。

クエリーのレイテンシのトラブルシューティングに関する推奨事項については、Kinesisのレイテンシを短縮するための推奨事項を参照してください。

認証

Kinesisは、Unity Catalog接続、サービス資格情報、またはインスタンスプロファイルやアクセスキーなどの代替メソッドによる認証をサポートしています。認証を参照してください。

スキーマ

Kinesis は次のスキーマでレコードを返します：

列	Type	説明
`partitionKey`	string	レコードが割り当てられているシャードを識別するパーティションキー。
`data`	binary	不透明なバイナリとしての、レコードのデータ BLOB。
`stream`	string	レコードの読み取り元となった Kinesis ストリームの名前または ARN。
`shardId`	string	レコードの読み取り元となったシャードの ID。
`sequenceNumber`	string	シャード内におけるレコードの一意の識別子。
`approximateArrivalTimestamp`	timestamp	レコードがストリームに挿入された概算時間。

列	Type	説明
`partitionKey`	string	レコードが割り当てられているシャードを識別するパーティションキー。
`data`	binary	不透明なバイナリとしての、レコードのデータ BLOB。
`stream`	string	レコードの読み取り元となった Kinesis ストリームの名前または ARN。
`shardId`	string	レコードの読み取り元となったシャードの ID。
`sequenceNumber`	string	シャード内におけるレコードの一意の識別子。
`approximateArrivalTimestamp`	timestamp	レコードがストリームに挿入された概算時間。

data 列のデータを逆シリアル化するには、フィールドを文字列にキャストします。

クイックスタート

次のノートブックは、Kinesisを使用した構造化ストリーミングを使用して WordCount を実行する方法を示しています。

Kinesis WordCount with Structured Streaming ノートブック

ノートブックを新しいタブで開く Open in Databricks

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 ソースを追加します。

Python
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 ストリームソースを削除するには、次の操作を行います。

クラスターの Spark構成で spark.databricks.kinesis.failOnDataLoss を false に設定し、クラスターを再起動します。failOnDataLoss の詳細については、データ損失の処理を参照してください。

streamName または streamARN オプションからストリームを削除し、クエリーを再起動します。たとえば、stream2 の読み取りを停止するには、次のようにします。

Python
df = (spark.readStream
  .format("kinesis")
  .option("streamName", "stream1,stream2")
  .load()
)

df.stop()

df = (spark.readStream
  .format("kinesis")
  .option("streamName", "stream1")  # Previous value was "stream1,stream2"
  .load()
)

spark.databricks.kinesis.failOnDataLoss を false に設定し、クラスターを1回再起動するだけで済みます。その後、そのクラスター上の追加のストリームソースを削除する場合、必要なのはクエリーの再起動のみであり、クラスターの再起動は不要です。

注記

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に設定すると、可能な限り頻繁にデータを取得します。
オプション shardsPerTask を設定するか、# cores in cluster >= 2 * (# Kinesis shards) / shardsPerTask となるようにクラスターを設定します。これにより、バックグラウンドのプリフェッチタスクとストリーミングクエリータスクが同時に実行されることが保証されます。

クエリーが 5 秒ごとにデータを受信している場合、Kinesis のレート制限を超える可能性があります。設定を見直してください。

Kinesis メトリクスの監視

Kinesisは、各ワークスペースごとにストリーム開始時に消費者が遅延するミリ秒数を報告します。avgMsBehindLatest、maxMsBehindLatest、minMsBehindLatestのメトリクスは、ストリーミングクエリープロセスのすべてのワークスペースにおける平均、最小、最大ミリ秒を提供します。Databricksの「Structured Streaming クエリーのモニタリング」を参照してください。

ノートブックでストリームを実行している場合は、ストリーミングクエリーの進行状況ダッシュボードの [ 生データ ] tabでメトリクスを確認してください。例を次に示します。

JSON
{
  "sources": [
    {
      "description": "KinesisV2[stream]",
      "metrics": {
        "avgMsBehindLatest": "32000.0",
        "maxMsBehindLatest": "32000",
        "minMsBehindLatest": "32000"
      }
    }
  ]
}

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：増分バッチ処理を参照してください。

データ損失の処理

警告

ワークロードがレコードの欠落を許容できる場合にのみ、failOnDataLoss を使用してください。これを誤って使用すると、データが完全に失われる可能性があります。レコードの欠落を許容できない場合は、新しいチェックポイントでストリームを再起動し、すべてのレコードを再処理してください。

Databricksでは、データ損失問題に対する一時的な緩和策としてのみこれを使用することを推奨しています。Kinesisの保持期間が短すぎるなど、根本原因を調査して修正します。

ストリーミングクエリーが読み取る前に Kinesis シャードのレコードが期限切れになった場合、または同じ名前で Kinesis ストリームを削除して再作成した場合、クエリーは KINESIS_COULD_NOT_READ_SHARD_UNTIL_END_OFFSET エラーで失敗します。KINESIS_COULD_NOT_READ_SHARD_UNTIL_END_OFFSET を参照してください。

defaultでは、ストリーミングクエリーは潜在的なデータ損失を検出すると失敗します。読み取り不可能なレコードをスキップして処理を継続するようにクエリーを構成するには、クラスターの Spark構成で spark.databricks.kinesis.failOnDataLoss を false に設定し、クラスターを再起動します。

Kinesis への書き込み

次のコードスニペットを、Kinesisにデータを書き込むためのForeachSinkとして使用します。これには Dataset[(String, Array[Byte])] が必要です。

注記

以下のコードスニペットは、「厳密に 1 回」ではなく、「 最低 1 回 」のセマンティクスを提供しています。

Kinesis Foreach シンクノートブック

ノートブックを新しいタブで開く Open in Databricks

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 シャードの利用率を最大化するため、問題ではない可能性があります。

レート制限エラーが多すぎることによる速度低下を回避

コネクタは、レート制限エラーが発生するたびにKinesisから読み取るデータ量を半分に減らし、このイベントをログにメッセージとして記録します： "Hit rate limit. Sleeping for 5 seconds."

ストリームが追いついている間にこれらのエラーが見られることがあります。ストリームが追いついた後にこれらのエラーが表示される場合は、AWS での Kinesis 容量を増やすか、Spark のプリフェッチオプションを調整してワークロードを調整する必要があるかもしれません。

ディスクへのspillを回避

Kinesis ストリームのデータ量が急増した場合、割り当てられたバッファ容量が一杯になり、新しいデータを追加するのに十分な速さで空き容量が確保されない可能性があります。Spark はバッファからディスクへデータをspillするため、ストリーム処理が遅延し、Logsに次のようなメッセージのイベントが表示されます：

Bash
./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を作成します。

Scala
import com.databricks.sql.kinesis.AWSKinesisConsumerManager

val manager = AWSKinesisConsumerManager.newManager()
.option("serviceCredential", serviceCredentialName)
.option("region", kinesisRegion)
.create()

コンシューマーをリストして表示します。

Scala
val consumers = manager.listConsumers("<stream name>")
display(consumers)

特定のストリームのコンシューマーを登録します。

Scala
val consumerARN = manager.registerConsumer("<stream name>", "<consumer name>")

特定のストリームのコンシューマーの登録を解除します。
Scala
```
manager.deregisterConsumer("<stream name>", "<consumer name>")
```

認証​

スキーマ​

クイックスタート​

Kinesis WordCount with Structured Streaming ノートブック

Kinesis オプションの設定​

ストリームソースの追加または削除​

ストリームの追加​

ストリームの削除​

低遅延のモニタリングとアラート​

Kinesis メトリクスの監視​

Kinesis レコードを増分バッチとして取り込む​

データ損失の処理​

Kinesis への書き込み​

Kinesis Foreach シンクノートブック

Kinesis のレイテンシーを削減するための推奨事項​

プリフェッチのレイテンシーを短縮​

レート制限エラーが多すぎることによる速度低下を回避​

ディスクへのspillを回避​

中断された S3 書き込みタスク​

ステートフルストリームでのチェックポイントによる遅延の削減​

ストリーミングクエリー読み取り用にKinesis Enhanced Fan-Out (EFO) を設定する​

Databricks ノートブックを使用したオフラインコンシューマー管理​

認証