既定では、パイプライン フローは Unity カタログによって管理される Delta テーブル (通常はテーブルまたは具体化されたビューをストリーミング) に結果を書き込みます。 シンクは、変換されたデータを Databricks マネージド ストレージの外部 (イベント ストリーミング サービスやカスタム データ ストアなど) の外部の宛先に書き込む代替出力ターゲットです。
シンクは追加フローで使用されます。 シンク API のいずれかを使用してシンクを定義し、それをtarget定義内のappend_flowとして参照します。
シンクを使用するタイミング
Databricks では、次の必要がある場合にシンクを使用することをお勧めします。
- クラウド ストレージではなくメッセージ バスにデータを流す必要がある、不正行為の検出、リアルタイム分析、顧客の推奨事項など、待機時間が短い運用ユース ケースを構築します。 ミリ秒の待機時間が必要なワークロードについては、「 Lakeflow パイプラインでリアルタイム モードを使用する」を参照してください。
- Unity カタログのマネージド テーブルや外部テーブルなど、外部 Delta インスタンスによって管理されるテーブルに変換されたデータを書き込みます。
- 処理されたデータを Apache Kafka トピックに書き戻して、Azure Databricks外で使用するなど、外部システムへの ETL の反転を実行します。
- Python カスタム データ ソースを使用して、Azure Databricks でネイティブにサポートされていない形式に書き込みます。
シンクの種類
パイプラインでは、次のシンクの種類がサポートされています。
| シンクの種類 | Description |
|---|---|
| デルタ テーブル シンク | Unity カタログのマネージド テーブルまたは外部 Delta テーブルに書き込みます。 ファイル パスまたは完全修飾テーブル名を指定します。 |
| Apache Kafka シンク | パイプライン ランタイムに含まれる Kafka コネクタを使用して Apache Kafka トピックに書き込みます。 |
| Azure Event Hubs シンク | Kafka インターフェイスを使用してAzure Event Hubsに書き込みます。 Kafka シンクと同じオプションを使用します。 |
| Python カスタムシンク |
spark.dataSource.registerに登録されているPythonカスタム データ ソースを使用して、任意のデータ ストアに書き込みます。 |
| ForEachBatch シンク | ストリーミング データの各マイクロバッチにカスタム Python ロジックを適用します。 複数の宛先への書き込み、アップサートの実行、またはストリーミング書き込みをネイティブでサポートしていないターゲットの使用が必要な場合に使用します。 |
シンクの API
パイプラインには、シンクを作成するための 2 つの API が用意されています。
-
create_sink(): サポートされている型 (Delta、Kafka、AEH、またはカスタム データ ソースPython) の名前付きシンクを作成します。 Pythonでのみ使用できます。 「パイプラインでシンクを使用する」を参照してください。 -
foreach_batch_sink():ストリーミング データのマイクロバッチごとに実行されるPython関数を装飾します。 カスタム書き込みロジックの柔軟性を最大限に高める。 「ForEachBatch を使用してパイプライン内の任意のデータ シンクに書き込む」を参照してください。
どちらのシンク型も、targetのappend_flowとして参照されます。
Limitations
- シンクは、Pythonでのみ使用できます。 SQL はサポートされていません。
- ストリーミング クエリのみがサポートされています。 バッチ クエリはサポートされていません。
- シンクに書き込むことができるのは
append_flowだけです。create_auto_cdc_flowおよびその他のフローの種類はサポートされていません。 - シンクでは、パイプラインの期待値はサポートされていません。
- 完全更新を実行しても、以前に書き込まれたシンク内のデータはクリーンアップされません。
