共有テーブルに対して受信者が実際に行うこと
によって ジョーダン・ソルド 、 Nikhil Tallam による投稿
多くのデータ共有アーキテクチャは、おなじみのパターンから始まります:
このブログのパート1では、定期的なファイルやAPIフィードの代わりに共有テーブルを使用することがなぜ優れた選択肢となるのかを説明しました。ここでは、受信側がそのテーブルを実際にどのように処理するのかを見ていきましょう。
このウォークスルーでは、Databricks間（D2D）の共有に焦点を当てます。受信側には Unity Catalog が有効化された Databricks ワークスペースがあり、共有カタログを介して共有テーブルにアクセスします。プロバイダーは履歴を含めてテーブルを共有するため、受信側はこれを Structured Streaming のソースとして使用できます。Databricksからオープン（D2O）への共有については、このブログの後半で改めて解説します。
Deltaテーブルには、行われたすべての変更の 順序付けられた履歴 が保持されます。OpenSharingはそのテーブルをコピーすることなく、オープンプロトコル経由で公開します。データはプロバイダーのストレージに置かれたまま、受信側がライブで読み取ります。
受信側では、Structured Streaming のリーダーを共有テーブルに向けます。Sparkはチェックポイントでどこまで読み込んだかを記録するため、各実行では前回からの変更分のみが取得されます。「前回確認した以降の新着情報」の追跡はエンジンの仕事となり、リネージやアクセス制御が維持された統制されたデータに対して実行されます。
例として、ある小売企業がライブの注文テーブルを維持し、Databricksを実行している外部の受信者とそれを共有しているケースを考えてみましょう。受信者は、毎夜CSVを受け取る代わりに、共有された注文テーブルを直接クエリできます。分析やダウンストリームのワークロード用にローカルコピーが必要な場合は、データを増分的にDeltaテーブルにマテリアライズ（実体化）できます。この設定は一度だけ行い、プロバイダー側と受信者側の両方で行います。
Change Data Feedを有効にしてソーステーブルを作成し、WITH HISTORYを指定して共有で公開します。履歴があることで、テーブルがインクリメンタルソースになります。履歴がない場合、受信側は静的なスナップショットしか取得できません。以下の主なインクリメンタル読み込みにおいて、Change Data Feed（delta.enableChangeDataFeed）の有効化はオプションです。後でアップサートや削除のパスを使用する予定がある場合のみ有効にしてください。Databricks Runtime 19以降を使用している場合は、このプロパティをスキップし、行トラッキングが有効なテーブルに対してクエリ時に行レベルの変更を計算し、Databricks-to-Databricks OpenSharingと連携する 自動Change Data Feed を利用できます。
共有識別子で受信者を登録し、共有を付与します:
CREATE SHARE および CREATE RECIPIENT には、ワークスペースの管理者だけでなく、メタスト アレベルの権限が必要であることに注意してください。
これで、メタストア内にプロバイダーとして共有が表示されます。カタログとしてマウントすれば、他のテーブルと同様に読み込むことができます。
共有が shared_in としてマウントされた状態で、読み込みは Structured Streaming によって処理されます。
Trigger.AvailableNow は、現在利用可能な未処理のデータを1つ以上のマイクロバッチを使用して処理し、その後終了します。これにより、スケジュールされたインクリメンタルジョブに適した動作になります。固定のスケジュールの代わりに、共有テーブル上で テーブル更新トリガー を使用してジョブを構成することもできるため、プロバイダーが新しいデータをコミットしたときのみジョブを実行できます。チェックポイントは UC Volumes などの耐久性のあるクラウドストレージに保存することが推奨されます。
注: この例ではDelta Sharingを介して共有テーブルを読み込むため、ignoreChangesを使用します。通常のDeltaテーブルから直接ストリーミングする場合、新しいDatabricksランタイムでは追記専用の処理に skipChangeCommits を使用します。
実際にインクリメンタルであることを証明するために、初期の行セットに対してストリームを1回実行します。プロバイダー側に新しい 行を追加した後、もう一度実行します。ターゲットテーブルのコミット履歴には、2回のストリーミング書き込みが表示されます:
バージョン | 操作 | 操作メトリクス |
|---|---|---|
2 | STREAMING UPDATE | numRemovedFiles: 0, numOutputRows: 200, numOutputBytes: 2307, numAddedFiles: 1 |
1 | STREAMING UPDATE | numRemovedFiles: 0, numOutputRows: 1000, numOutputBytes: 4958, numAddedFiles: 1 |
0 | CREATE TABLE | {} |
2回目の実行で書き込まれたのは1200行ではなく200行です。チェックポイントで前回読み込んだバージョンが追跡されているため、今回の実行では新しい行のみが取得されました。
ignoreChangesはこのインクリメンタルランディングパターンに適していますが、行が再発行される可能性があり、削除は伝播されません。ダウンストリームのコンシューマーが挿入、更新、削除を必要とする場合は、代わりに共有テーブルのChange Data Feedを読み込んでください:
新しい顧客が追加された場合でも、プロバイダー側で新しいSFTPエンドポイントを作成したり、新しいエクスポートジョブを実装したりする必要はありません。プロバイダーは受信者に対して共有へのアクセス権を付与します。受信者は、独自のターゲットテーブルや下流のビジネスロジックを分離したまま、同じ一般的なパターンを使用します。運用上の違いは非常に大きいです:
上記の内容がすべて機能するのは、受信者がDatabricks上でSparkを実行しており、それがチェックポイント付きストリーミングとCDFを可能にしているためです。その他の受信者（pandas、BIツール、Snowflakeなど）は、メタストアの共有IDの代わりに認証情報ファイルを使用してOpenSharing経由で接続します。
OSS Sparkを実行している場合、読み込みは delta-sharing コネクタを通した同じ spark.readStream コードになります。Sparkがない場合は継続的なストリームが存在しないため、バージョン範囲を指定して変更を取得します:
有用な最初のプロジェクトでは、すべての共有シナリオを解決する必要はありません。1つの通常のDeltaテーブルと1人の受信者から始めてください。
ストリームを実装する前に、以下の質問に答えてください:
これらの回答が明確になったら、可能な限り最小限のエンドツーエンドのフローを検証します。共有の作成、初期スナップショットの読み込み、新しい行の書き込み、行の更新、行の削除を行い、受信者が期待通りの結果を確認できることを確認します。
これが、OpenSharingによって可能になる実用的な転換です。顧客ごとに新しいトランスポートや取り込みワークフローを構築する代わりに、ガバナンスされたデータ共有契約を確立し、各受信者がそのプラットフォームに適したパスを通じてデータを消費できるようにすることができます。
OpenSharing を初めてご利用ですか？OpenSharing とは？ と OpenSharing 用の共有を作成する から始めて、上記の最小限のエンドツーエンドのフローを 1 つのテーブルと 1 つの受信者でお試しください。
(このブログ記事はAI翻訳ツールを使用して翻訳されています) 原文記事
ブログを購読して、最新の投稿を受信トレイにお届けします。