DLT Sink API のご紹介：Kafka および外部 Delta テーブルへのパイプライン書き込み

Delta Live Tablesを初めてお使いになる方は、このブログを読む前に、Delta Live Tables (DLT) の宣言的なETL定義とステートメントを使用して、スケーラブルで信頼性の高いパイプラインを構築する方法を説明しているDelta Live Tablesの始め方をお読みになることをお勧めします。

はじめに

Delta Live Tables (DLT)パイプラインは、Databricks内で信頼性が高く、保守可能で、テスト可能なデータ処理パイプラインを構築するための堅牢なプラットフォームを提供します。宣言的なフレームワークを活用し、最適なサーバーレスコンピューティングを自動的にプロビジョニングすることで、DLTはストリーミング、データ変換、管理の複雑さを簡素化し、最新のデータワークフローにスケーラビリティと効率性をもたらします。

従来、DLTパイプラインは、Unity Catalogによって管理されるストリーミングテーブルまたはマテリアライズドビューとして、効率的にデータをインジェストおよび処理する方法を提供してきました。このアプローチはほとんどのデータ処理ニーズを満たしますが、データパイプラインが外部システムと連携する必要がある場合や、ストリーミングテーブルまたはマテリアライズドビューに書き込む代わりにStructured Streamingシンクを使用する必要がある場合があります。

DLTの新しいシンクAPIの導入は、ユーザーが処理済みデータをApache KafkaやAzure Event Hubsなどの外部イベントストリームに書き込んだり、Deltaテーブルに書き込んだりできるようにすることで、この問題に対処します。この新しい機能により、DLTパイプラインの範囲が広がり、外部プラットフォームとのシームレスな統合が可能になります。

これらの機能は現在パブリックプレビュー中で、今後、Databricks RuntimeからDLTへのシンクを追加し、最終的にはすべてをサポートする予定です。現在取り組んでいる次の機能は、foreachBatchであり、これにより顧客は任意のデータシンクに書き込み、Deltaテーブルへのカスタムマージを実行できます。

Sink APIはdlt Pythonパッケージで利用可能で、以下に示すようにcreate_sink()で使用できます。

APIは、シンクを定義する際に3つの主要な引数を受け取ります。

• シンク名: パイプライン内でシンクを一意に識別する文字列。この名前により、シンクを参照および管理できます。
• フォーマット指定: 出力フォーマットを決定する文字列で、「kafka」または「delta」をサポートします。
• シンクオプション: キーと値のペアの辞書で、キーと値の両方が文字列です。Kafkaシンクの場合、Structured Streamingで利用可能なすべての設定オプション（認証、パーティショニング戦略などの設定を含む）を利用できます。Kafkaでサポートされている設定オプションの包括的なリストについては、ドキュメントを参照してください。Deltaシンクは、path属性を使用してストレージパスを定義するか、tableName属性を使用してUnity Catalogのテーブルに直接書き込むことで、よりシンプルな設定を提供します。

シンクへの書き込み

@append_flow APIは、シンク名で識別されるターゲットシンクにデータを書き込むことができるように拡張されました。従来、このAPIは、ユーザーが複数のソースから単一のストリーミングテーブルにシームレスにデータをロードすることを可能にしていました。新しい拡張機能により、ユーザーは特定のシンクにもデータを追加できるようになりました。以下は、この設定方法を示す例です。

パイプラインの構築

次に、Databricksのデータセットに含まれるクリックストリームデータを処理するDLTパイプラインを構築しましょう。このパイプラインは、データを解析してApache Sparkページへのイベントを特定し、そのデータをEvent HubsとDeltaシンクの両方に書き込みます。品質と処理効率を向上させるためにデータを異なるレイヤーに整理するMedallionアーキテクチャを使用してパイプラインを構造化します。

まず、Auto Loaderを使用して生のJSONデータをBronzeレイヤーにロードします。次に、Silverレイヤーでデータをクリーニングし、品質基準を適用して整合性を確保します。最後に、Goldレイヤーで、現在のページタイトルがApache_Sparkのエントリをフィルタリングし、spark_referrersという名前のテーブルに保存します。これはシンクのソースとして機能します。完全なコードについては、付録を参照してください。

Azure Event Hubsシンクの設定

このセクションでは、create_sink APIを使用してEvent Hubsシンクを確立します。これは、運用中のKafkaまたはEvent Hubsストリームがあることを前提としています。パイプラインは、共有アクセス ポリシーを使用してKafka対応のEvent Hubsにデータをストリーミングし、接続文字列はDatabricks Secretsに安全に保存されます。または、SASポリシーの代わりにサービスプリンシパルを使用して統合することもできます。接続プロパティとシークレットを適切に更新してください。以下は、Event Hubsシンクを設定するためのコードです。

Deltaシンクの設定

Event Hubsシンクに加えて、create_sink APIを使用してDeltaシンクを設定することもできます。このシンクは、Databricks File System (DBFS) の指定された場所にデータを書き込みますが、Amazon S3やADLSなどのオブジェクトストレージの場所に書き込むように設定することもできます。

以下は、Deltaシンクを設定する方法の例です。

KafkaおよびDeltaシンクをデータで満たすためのフローの作成

Event HubsおよびDeltaシンクが確立されたら、次のステップはappend_flowデコレータを使用してこれらのシンクをデータで満たすことです。このプロセスには、シンクにデータをストリーミングすることが含まれ、最新の情報で継続的に更新されることが保証されます。

Event Hubsシンクの場合、valueパラメータは必須ですが、key、partition、headers、topicなどの追加パラメータはオプションで指定できます。以下は、KafkaおよびDeltaシンクの両方のフローを設定する方法の例です。

applyInPandasWithState関数もDLTでサポートされるようになり、ユーザーはPandasのパワーを活用してDLTパイプライン内でステートフルな処理を行うことができます。この拡張機能により、使い慣れたPandas APIを使用して、より複雑なデータ変換と集計が可能になります。DLT Sink APIを使用すると、ユーザーはこのステートフルに処理されたデータをKafkaトピックに簡単にストリーミングできます。この統合は、リアルタイム分析およびイベント駆動型アーキテクチャに特に役立ち、データパイプラインがストリーミングデータを外部システムに効率的に処理および配信できるようにします。

まとめ

上記のアプローチは、DLTパイプラインを構築してデータを効率的に変換し、新しいSink APIを使用して結果を外部DeltaテーブルおよびKafka対応Event Hubsにシームレスに配信する方法を示しています。

この機能は、リアルタイム分析パイプラインにとって特に価値があり、異常検出、予知保全、その他の時間的制約のあるユースケースなどのアプリケーションのために、データをKafkaストリームにストリーミングできます。また、イベント駆動型アーキテクチャも可能になり、ダウンストリームプロセスはKafkaトピックへのストリーミングイベントによって即座にトリガーされ、新しく到着したデータの高速処理が可能になります。

Call to Action

DLT Sinks 機能が、すべての Databricks お客様向けにパブリックプレビューで利用可能になりました！この強力な新機能により、Kafka や Delta Lake テーブルなどの外部システムに DLT パイプラインをシームレスに拡張でき、リアルタイムのデータフローと合理化された統合が保証されます。詳細については、以下のリソースを参照してください。

• DLT Sinks に関する Databricks ドキュメント
• Confluent Cloud を使用した DLT での Kafka Sinks に関する デモを視聴
• DLT の Python 言語リファレンス

Appendix:

Pipeline Code:

(このブログ記事はAI翻訳ツールを使用して翻訳されています) 原文記事