Unity CatalogでのDelta Sharingを使用したStructured Streamingの使用
によって Will Girten, Josh Seidel, Lin Zhou 、 サチン タクール による投稿
Delta SharingでのStructured Streamingのサポートが、Azure、AWS、GCPで一般公開（GA）されたことをお知らせします。この新機能により、Databricks Lakehouse Platform上のデータ受信者は、Unity Catalogを通じて共有されたDeltaテーブルからの変更をストリーミングできるようになります。
データ提供者はこの機能を活用することで、Data-as-a-Serviceを容易に拡張し、大規模データセットの共有に伴う運用コストを削減できるほか、新しいデータが届いた際に即座にデータ検証や品質チェックを行ってデータ品質を向上させたり、リアルタイムのデータ配信によってカスタマーサービスを向上させたりできます。同様に、データ受信者は共有データセットから最新の変更をストリーミング処理できるため、大規模なバッチデータ の処理にかかるインフラコストを削減し、最先端のリアルタイムデータアプリケーションの基盤を構築できます。次のように、多くの業界領域のデータ受信者がこの新機能の恩恵を受けることができます。
- 小売：データアナリストは、季節限定ファッションラインの最新売上数値をストリーミングし、BIレポート形式でビジネスインサイトを提示できます。
- ヘルスケア・ライフサイエンス：医療従事者は、心電図の測定値をMLモデルにストリーミングして異常を識別できます。
- 製造：ビル管理チームはスマートサーモスタットの測定値をストリーミングし、冷暖房ユニットを効率的にオン/オフすべき時間帯（昼夜問わず）を特定できます。
バッチ実行は堅牢で実装が容易であるため、データチームはデータの処理をバッチ方式で実行されるデータパイプラインに依存することが多くあります。しかし今日、組織がリアルタイムでビジネスの意思決定を行うには、最新のデータが必要です。Structured Streamingはリアルタイム処理を簡素化するだけでなく、多数のバッチジョブをわずかなストリーミングジョブに削減することで、バッチ処理も簡素化します。Structured Streamingは同じDataFrame APIをサポートしているため、バッチデータパイプラインからストリーミングへの変換は非常に簡単です。
このブログ記事では、金融業界の例を用いて、企業がDelta SharingとともにStructured Streamingを活用し、ニアリアルタイムでデータのビジネス価値を最大限に高める方 法を探ります。また、Databricks Workflowsなどの補完的な機能をDelta SharingやUnity Catalogと組み合わせて使用し、リアルタイムデータアプリケーションを構築する方法についても検証します。
Structured Streamingのサポート
過去数か月間で最も高く期待されていたDelta Sharing機能の1つは、共有されたDeltaテーブルをStructured Streamingのソースとして使用するサポートの追加でしょう。この新機能により、データ受信者はDatabricks Lakehouse Platform上でUnity Catalogを通じて共有されたDeltaテーブルを使用してリアルタイムアプリケーションを構築できるようになります。
Structured StreamingでDelta Sharingを使用する方法
リアルタイムのトレードインサイトを得るために、データ受信者が上場株式のシンボル情報をどのようにストリーミングできるかを詳しく見ていきましょう。この記事では、米国全米市場システム （NMS）で取引されているすべての株式および持分証券を一覧化したFINRA CAT Reportable Equity Securities Symbol Masterデータセットを使用します。Structured Streamingはリアルタイムアプリケーションの構築に使用できますが、データの到着頻度が低いシナリオでも有用です。簡単なノートブックのデモとして、1日に3回更新されるデータセットを使用します。1回目は取引日の開始時（SOD）、2回目は日中の変更を反映する日中、3回目は取引日の終了時（EOD）です。週末や米国の祝日には更新は公開されません。
| 公開ファイル | スケジュール |
|---|---|
| CAT Reportable Equity Securities Symbol Master – SOD | 午前6:00（EST） |
| CAT Reportable Options Securities Symbol Master – SOD | 午前6:00（EST） |
| Member ID (IMID) List | 午前6:00（EST） |
| Member ID (IMID) Conflicts List | 午前6:00（EST） |
| CAT Reportable Equity Securities Symbol Master – Intraday | 午前10:30（EST）、およびEODファイルが公開されるまで約2時間ごと |
| CAT Reportable Options Securities Symbol Master – Intraday | 午前10:30（EST）、およびEODファイルが公開されるまで約2時間ごと |
| CAT Reportable Equity Securities Symbol Master – EOD | 午後8:00（EST） |
| CAT Reportable Options Securities Symbol Master – EOD | 午後8:00（EST） |
表1.1 - FINRA CATシンボルおよびメンバーのリファレンスデータは、営業日を通じて公開されます。週末や米国の祝日には更新は公開されません。
データ提供者の視点：Databricks Workflowsを使用したCATデータの取り込み
Databricks Lakehouse Platformの大きな利点の1つは、変更をDeltaテーブルへ継続的にストリーミングするのが極めて簡単であることです。まず、取引日の開始時（SOD）にFINRA CAT株式シンボルファイルをダウンロードする簡単なPythonタスクを定義することから始めます。その後、公開されたファイルをDatabricksファイルシステムの臨時ディレクトリに保存します。
コード 1.1 - 簡単なPythonタスクにより、取引日の開始時にFINRA CAT株式シンボルファイルをダウンロードできます。
デモとして、RAWファイルを取り込み、更新されたファイルが公開されるたびにDelta Lake内のブロンズテーブルを継続的に更新する関数も定義します。
コード 1.2 - FINRA CAT株式シンボルデータは、各取引日の開始時にDeltaテーブルに取り込まれます。
一度開始されると、Databricks Workflowは取引日の開始時にファイルが公開されるたびに、CAT株式シンボルデータセットへのデータ投入を開始します。
データ提供者の視点：Deltaテーブルをストリーミングソースとして共有
各取引日にシンボルファイルへの更新を取り込むストリーミングパイプラインを作成できたので、Delta Sharingを活用してデータ受信者とDelta Tableを共有できるようになりました。DatabricksレイクハウスプラットフォームでのDelta Shareの作成は、ボタンを数回クリックするか、SQL構文を好む場合は1つのSQL文を実行するだけで行えます。
同様に、データプロバイダーは［アセットの管理］ボタンをクリックし、続いて［テーブルの編集］ボタンをクリックすることで、1つ以上のテーブルをDelta Shareに追加できます。この例では、株式シンボルデータを含むブロンズのDelta TableがShareオブジェクトに追加されます。
Structured Streamingを使った読み取りをサポートするには、Deltaテーブルの完全な履歴を共有する必要があることに注意してください。Databricks UIを使用してDeltaテーブルをShareに追加する場合、履歴の共有はデフォルトで有効になっています。ただし、SQL構文を使用する場合は、履歴の共有を明示的に指定する必要があります。
コード 1.4 - SQL構文を使用する場合、Structured Streamingの読み取りをサポートするにはDeltaテーブルの履歴を明示的 に共有する必要があります。
データ受信者からの視点：共有されたDelta Tableのストリーミング
データ受信者としても、共有されたDeltaテーブルからのストリーミングは非常に簡単です。Delta Shareがデータ受信者と共有されると、受信者のUnity Catalog内のプロバイダー詳細の下にそのShareがすぐに表示されます。その後、データ受信者は［カタログを作成］ボタンをクリックし、わかりやすい名前を入力して、必要に応じてShareの内容を説明するコメントを追加することで、Unity Catalog内に新しいカタログを作成できます。
データ受信者は、Databricks Runtime 12.1以降を使用して、Unity Catalogを通じて共有されたDelta Tableからストリーミングできます。この例では、Databricks 12.2 LTS RuntimeがインストールされたDatabricksクラスターを使用しています。データ受信者は、deltaSharingデータソースを使用し、共有テーブルの名前を指定することで、共有DeltaテーブルをSpark Structured Streamとして読み取ることができます。
コード 1.4 - データ受信者はdeltaSharingデータソースを使用して共有Delta Tableからストリーミングできます。
さらに別の例として、共有されているCAT株式シンボルのマスターデータセットと、データ受信者のローカルUnity Catalogにある株価履歴データセットを組み合わせてみましょう。まず、指定された銘柄コードの週次株価履歴を取得するためのユーティリティ関数を定義します。
次に、株式マスターデータのストリームと、大手IT企業3社（Apple Inc. (AAPL)、Microsoft Corporation (MSFT)、Invidia Corporation (NVDA)）のローカル株価履歴を結合します。
最後に、データ受信者はオプションで送信先シンクを追加し、ストリーミングクエリを開始できます。
DatabricksでDelta Sharingをはじめる
組織がDelta Sharingを活用して、ほぼリアルタイムでデータのビジネス価値を最大化する方法についてのこの例をお楽しみいただけたでしょうか。
Delta Sharingを始めたいものの、どこから手をつければよいかわからないですか？すでにDatabricksをご利用のお客様は、ガイドに従ってDelta Sharingを開始してください（AWS | Azure | GCP）。この機能に含まれる設定オプションの詳細については、ドキュメントをご覧ください。まだDatabricksをご利用でない場合は、PremiumまたはEnterpriseワークスペースの無料トライアルにお申し込みください。
クレジット
Abhijit Chakankar、Lin Zhou、Shixiong Zhuをはじめとする、本リリースへのすべての貢献に特別に感謝申し上げます。
リソース
(このブログ記事はAI翻訳ツールを使用して翻訳されています) 原文記事
最新の投稿を受信トレイで受け取る
ブログを購読して、最新の投稿を受信トレイにお届けします。