Sinks in LakeFlow Pipelines
デフォルトでは、パイプラインフローは、Unity Catalog が管理する Delta テーブル (通常、ストリーミングテーブルまたはマテリアライズドビュー) に結果を書き込みます。シンクは、イベントストリーミングサービスやカスタムデータストアなど、Databricks が管理するストレージの外部にある宛先に、変換されたデータを書き込むために使用する代替ターゲットです。
sink APIのいずれかを使用してシンクを定義してから、append_flowまたはupdate_flow定義内でそれをtargetとして参照します。
シンクを使うタイミング
Databricksでは、次の要件がある場合にシンクを使用することをお勧めします：
- 詐欺検出、リアルタイムアナリティクス、顧客レコメンデーションなど、データがクラウドストレージではなくメッセージバスに流れる必要がある低レイテンシーの運用ユースケースを構築します。ミリ秒単位のレイテンシーを必要とするワークロードについては、LakeFlow Pipelinesでリアルタイムモードを使用するを参照してください。
- 変換されたデータを、外部Deltaインスタンスによって管理されるテーブル (Unity Catalog マネージドテーブルや外部テーブルなど) に書き込みます。
- Databricks の外部での利用のために処理済みデータを Apache Kafka トピックに書き戻すといった外部システムへのリバース ETL を行います。
- Databricksでネイティブにサポートされていない形式に、Pythonカスタムデータソースを使用して書き込みます。
シンクのタイプ
パイプラインは次のシンクの種類をサポートしています。
シンクタイプ | 説明 |
|---|---|
Delta テーブル シンク | Unity Catalog の管理型または外部 Delta テーブルに書き込むことができます。ファイルパス、または完全修飾テーブル名を指定してください。 |
Apache Kafka シンク | パイプラインランタイムに含まれる Kafka コネクタを使用して、Apache Kafka トピックに書き込みます。 |
Azure Event Hubs シンク | Kafka インターフェースを使用した Azure Event Hubs への書き込みKafka シンクと同じオプションを使用します。 |
Pythonカスタムシンク | Pythonカスタムデータソースを使用して、 |
| ストリーミングデータの各マイクロバッチにカスタムPythonロジックを適用できます。複数の宛先へ書き込む必要がある場合、アップサートを実行する場合、またはストリーミング書き込みをネイティブでサポートしていないターゲットを使用する場合に利用します。 |
| ストリーミングデータの各行にカスタムPythonロジックを適用します。外部APIやキー値ストアなど、組み込みシンクがない宛先に各レコードを書き込む必要がある場合に使用します。 |
シンク APIs
パイプラインには、シンクを作成するための3つのAPIsがあります。
create_sink(): サポートされている種類（Delta、Kafka、AEH、またはPythonカスタムデータソース）の名前付きシンクを作成します。Pythonのみで利用可能です。「パイプラインでシンクを使用する」を参照してください。foreach_batch_sink()：ストリーミングデータの各マイクロバッチで実行されるPython関数を装飾します。カスタムロジックの記述に最大限の柔軟性を提供します。「ForEachBatch を使用してパイプライン内の任意のデータシンクに書き込む」を参照してください。foreach_sink(): Python関数をデコレートするか、ストリーミングデータの各行に対して実行されるライターオブジェクトを受け取ります。ForEachシンクを使用してパイプライン内の各行に書き込むを参照してください。
3つのシンクタイプすべてを append_flow または update_flow の target として参照します。
制限事項：
-
シンクはPythonでのみ使用できます。SQL はサポートされていません。
-
ストリーミングクエリのみがサポートされています。バッチクエリはサポートされていません。
-
シンクへの書き込みには
append_flowとupdate_flowのみを使用できます。create_auto_cdc_flowおよびその他のフロー タイプはサポートされていません。- アップデートフローは、Delta Lakeテーブルシンクをサポートしていません。
-
パイプラインの期待値はシンクではサポートされていません。
-
フル更新を実行しても、シンク内の以前に書き込まれたデータはクリーンアップされません。