新登場！DLT Sink APIでKafkaと外部Deltaテーブルにデータ書き込み

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

はじめに

 

Delta Live Tables (DLT)は、Databricks上で信頼性が高く、保守性とテスト性に優れたデータ処理パイプラインを構築できる強力なプラットフォームです。宣言型フレームワークを活用し、最適なサーバーレスコンピュートを自動的にプロビジョニングすることで、ストリーミング処理やデータ変換、管理の複雑さを大幅に軽減し、スケーラビリティと効率性を実現します。

これまで、DLTパイプラインは、Unity Catalogで管理されるストリーミングテーブルまたはマテリアライズドビューとしてデータを取り込み、処理する方法を提供してきました。このアプローチは多くのデータ処理ニーズに対応できますが、一部のユースケースでは、外部システムと連携する必要がある、またはStructured StreamingのSinkを活用したいといった要件が生じることがあります。

この課題を解決するために、新たにDLTのSinks APIが導入されました！これにより、Apache KafkaやAzure Event Hubsなどの外部イベントストリームへのデータ書き込み、さらには外部のDeltaテーブルへのデータ書き込みが可能になります。この新機能により、DLTパイプラインの適用範囲が広がり、外部プラットフォームとのシームレスな統合が実現します。

この機能は現在Public Previewとして提供されており、今後Databricks RuntimeのさらなるSinkサポートをDLTに追加していく予定です。次に開発を進めているのはforeachBatchで、これにより任意のデータSinkへの書き込みや、Deltaテーブルへのカスタムマージが可能になります。

Sink APIの利用方法
この新しいSink APIは、dlt Pythonパッケージ内で提供されており、以下のように**create_sink()**を使用して活用できます：

Sinkを定義するためにAPIが受け取る主要な3つの引数

1. Sink Name: パイプライン内でSinkを一意に識別する文字列。この名前を使用して、Sinkを参照および管理できます。
2. Format Specification: 出力形式を決定する文字列で、"kafka"または"delta"のいずれかを指定できます。
3. Sink Options: キーと値が文字列である辞書型のペア。Kafka Sinkの場合、Structured Streamingで利用可能なすべての設定オプション（認証設定、パーティション戦略など）を活用できます。Kafkaに対応した設定オプションの詳細については、ドキュメントを参照してください。Delta Sinkは、path属性を使用してストレージパスを定義するか、tableName属性を使用してUnity Catalog内のテーブルに直接書き込むことで、よりシンプルな構成を提供します。

Sinkへのデータ書き込み

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

パイプラインの構築

ここでは、Databricksのデータセットを活用してクリックストリームデータを処理するDLTパイプラインを構築します。このパイプラインは、データを解析し、Apache Sparkのページにリンクするイベントを識別した後、Event HubsとDelta Sinkの両方にデータを書き込むように設計されています。

このパイプラインでは、メダリオンアーキテクチャを採用し、データを異なるレイヤーに整理することで、データの品質向上と処理効率の最適化を図ります。

1. Bronzeレイヤー: Auto Loaderを使用して、JSON形式の生データをロード。
2. Silverレイヤー: データのクレンジングと品質チェックを行い、データの整合性を確保。
3. Goldレイヤー: Apache_Sparkという現在のページタイトルを持つエントリをフィルタリングし、spark_referrersというテーブルに保存。このテーブルが、Sinkへのデータ出力元となります。

完全なコードについては、付録を参照してください。

Azure Event Hubs Sinkの設定

このセクションでは、create_sink APIを使用してEvent HubsのSinkを構築します。ここでは、すでにKafkaまたはEvent Hubsのストリームが稼働していることを前提とします。

パイプラインは、Kafka対応のEvent Hubsにデータをストリーミングし、Databricks Secretsに安全に保存された共有アクセスポリシーの接続文字列を使用します。また、SASポリシーの代わりにサービスプリンシパルを利用する方法も可能です。接続プロパティやシークレットの設定を適宜更新してください。

以下は、Event Hubs Sinkを設定するためのコードです：

Delta Sinkの設定

Event Hubs Sinkに加えて、create_sink API を使用してDelta Sink を設定することも可能です。このSinkは、Databricks File System（DBFS）の指定された場所にデータを書き込むように設定できますが、Amazon S3やAzure Data Lake Storage（ADLS）などのオブジェクトストレージにも書き込むように構成できます。

以下は、Delta Sinkを設定するコードの例です：

KafkaおよびDelta Sinkへのデータフローの作成

Event Hubs（Kafka）SinkとDelta Sinkの設定が完了したら、次のステップとして**append_flow デコレーターを使用してこれらのSinkにデータを供給（ハイドレート）**します。このプロセスでは、データをストリーミングし、常に最新の情報で更新されるようにします。

Event Hubs（Kafka）Sinkの場合、value パラメーターが必須となりますが、key、partition、headers、topic などの追加パラメーターもオプションで指定できます。

以下に、Kafka（Event Hubs）SinkとDelta Sinkの両方にデータをストリーミングするためのフローの設定例を示します：

DLTでは、applyInPandasWithState 関数のサポートが追加され、Pandasの機能を活用したステートフルなデータ処理が可能になりました。この強化により、Pandas APIを使って、より複雑なデータ変換や集計処理をDLTパイプライン内で実装できます。さらに、DLT Sink APIを活用することで、このステートフル処理済みデータをKafkaトピックにストリーミングすることが可能になります。

この統合は、リアルタイム分析やイベント駆動型アーキテクチャに特に有用であり、データパイプラインが外部システムに対して効率的にストリーミングデータを処理・分配できる環境を提供します。

すべてを統合：DLT Sink APIによる効率的なデータ配信

ここまで紹介したアプローチでは、DLTパイプラインを構築し、新しいSink APIを活用して外部のDeltaテーブルやKafka対応のEvent Hubsへデータをシームレスに配信する方法を紹介しました。

この機能は特にリアルタイム分析パイプラインにおいて価値が高く、異常検知（Anomaly Detection）や予測保全（Predictive Maintenance）などの時間依存のユースケースに対応するために、データをKafkaストリームに送信できます。

また、イベント駆動型アーキテクチャにも対応し、Kafkaトピックへのストリーミングイベントによってダウンストリームプロセスを即座にトリガーできるため、新たに到着したデータを高速で処理できるようになります。

アクションの呼び出し

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

• DLTシンクについてのDatabricksドキュメント
• デモを見る Kafka Sinks in DLT using Confluent Cloud
• DLTのPython言語リファレンス

付録

パイプラインコード：