同期されたテーブルを使用して Lakehouse データを提供する

同期されたテーブルを使用すると、Lakebase Postgres を介して Lakehouse データを提供できます。 Unity カタログ テーブルは Postgres に同期されるため、アプリケーションは低待機時間で lakehouse データに直接クエリを実行できます。 このプロセスは、一般に逆 ETL と呼ばれます。 Lakehouse は分析とエンリッチメント用に最適化されていますが、Lakebase は高速なルックアップ スタイルのクエリとトランザクションの一貫性を必要とする運用ワークロード向けに設計されています。

同期テーブルとは

同期されたテーブルを使用すると、Unity カタログから Lakebase Postgres を介して分析グレードのデータを提供できるため、待機時間の短いクエリと完全な ACID トランザクションを必要とするアプリケーションで使用できます。 リアルタイム アプリケーションでデータを提供する準備を整えることで、分析ストレージと運用システムのギャップを埋めます。

サポートされているソース

同期テーブルでは、次の Unity カタログ ソースの種類がサポートされます。

• 管理されたデルタ テーブルと外部デルタ テーブル
• 管理されたIcebergテーブルと外部Icebergテーブル
• ビューと具体化されたビュー

どのように機能するのか

Databricks 同期テーブルは 、Lakebase に Unity カタログ データのマネージド コピーを作成します。 同期されたテーブルを作成すると、次の情報が得られます。

1. 同期パイプラインを参照する Unity カタログの同期テーブル
2. Lakebase の Postgres テーブル (アプリケーションで読み取り専用、クエリ可能)

たとえば、ゴールド テーブル、エンジニアリングされた機能、または analytics.gold.user_profiles からの ML 出力を新しい同期テーブル analytics.gold.user_profiles_syncedに同期できます。 Postgres では、Unity カタログ スキーマ名が Postgres スキーマ名になるため、 gold.user_profiles_syncedとして表示されます。

SELECT * FROM gold.user_profiles_synced WHERE user_id = 12345;

アプリケーションは標準の Postgres ドライバーに接続し、同期されたデータを独自の運用状態と共に照会します。

同期パイプラインでは、マネージド Lakeflow Spark 宣言パイプラインを使用して、Unity カタログ同期テーブルと Postgres テーブルの両方をソース テーブルからの変更で継続的に更新します。 各同期では、Lakebase データベースへの最大 16 個の接続を使用できます。

Lakebase Postgres では、トランザクションが保証された最大 1,000 個の同時接続がサポートされているため、アプリケーションはエンリッチされたデータを読み取り、同じデータベース内の挿入、更新、削除も処理できます。

同期モード

アプリケーションのニーズに基づいて適切な同期モードを選択します。

トリガーモードと連続モードでは、ソース テーブルで Change Data Feed (CDF) を有効にする必要があります。 CDF が有効になっていない場合は、実行する正確な ALTER TABLE コマンドを使用して UI に警告が表示されます。 変更データ フィードの詳細については、 Databricks での Delta Lake 変更データ フィードの使用に関するページを参照してください。

利用事例の例

同期テーブルは、次のようなデータサービスのユース ケースに使用できます。

• 新しいユーザー プロファイルを Databricks Apps に提供するパーソナル化エンジン
• Lakehouse で計算されたモデル予測または特徴値を提供するアプリケーション
• リアルタイムで KPI を提供する顧客向けのダッシュボード
• リスク スコアを即時アクションに提供する不正行為検出サービス
• Lakehouse データから強化された顧客レコードを提供する支援ツール

同期テーブルを作成する

前提条件

必要なもの:

• Lakebase が有効になっている Databricks ワークスペース。
• Lakebase プロジェクト (プロジェクト の作成を参照)。
• 同期する Unity カタログ テーブル。
• 同期されたテーブルを作成するためのアクセス許可。 使用するスキーマ にUSE_SCHEMA と CREATE_TABLE が必要です。

トリガーモードまたは連続モードの場合、ソース テーブルで Change Data Feed を有効にする必要があります。

ALTER TABLE your_catalog.your_schema.your_table
SET TBLPROPERTIES (delta.enableChangeDataFeed = true)

容量計画とデータ型の互換性については、「データ型と互換性」および「容量計画」を参照してください。

UI

1. ワークスペースのサイドバーで [カタログ ] に移動し、同期する Unity カタログ テーブルを選択します。

   

2. テーブルの詳細ビューで[ 作成>同期テーブル ]をクリックします。

   

3. [同期テーブルの作成] ダイアログで、次 の手順を実行 します。

   カタログとスキーマの一覧には、現在のユーザーが USE_SCHEMA 権限と CREATE_TABLE 権限を持つ Unity カタログ スキーマのみが含まれます。 予想されるスキーマが表示されない場合は、カタログ管理者にアクセス許可を確認してください。

   1. テーブル名: 同期されたテーブルの名前を入力します (ソース テーブルと同じカタログとスキーマに作成されます)。 これにより、Unity カタログ同期テーブルと、クエリできる Postgres テーブルの両方が作成されます。

   2. データベースの種類: Lakebase サーバーレス (自動スケール) を選択します。

   3. 同期モード: ニーズに基づいて [スナップショット]、[ トリガー済み]、または [継続的 ] を選択します (上記の 同期モード を参照)。

   4. プロジェクト、ブランチ、データベースの選択を構成します。

   5. 主キーが正しいことを確認します (通常は自動検出されます)。

      Important

      同期されたテーブルでは、主キーの列は null 許容ではありません。 主キー列に null が含まれる行は、 同期から除外されます。

   6. (省略可能)ソース テーブルで 2 つの行が同じ主キーを共有できる場合は、重複除去を構成する 時系列キー を選択します。 時系列キーを指定すると、同期されたテーブルには、各主キーの最新の時系列キー値を持つ行のみが含まれます。 時系列キーのない障害モードについては、「キーの複製」 を参照してください。

   トリガー モードまたは連続モードを選択し、データ フィードの変更をまだ有効にしていない場合は、正確なコマンドを実行すると警告が表示されます。 データ型の互換性に関する質問については、「 データ型と互換性」を参照してください。

   [ 作成 ] をクリックして、同期されたテーブルを作成します。

4. カタログで同期されたテーブルを監視します。 [ 概要 ] タブには、同期の状態、構成、パイプラインの状態、最後の同期タイムスタンプが表示されます。 [ 今すぐ同期] を使用して手動で更新します。

CLI

databricks postgres create-synced-table my-catalog.sales.orders \
  --json '{
    "spec": {
      "source_table_full_name": "main.sales.orders",
      "branch": "projects/my-project/branches/production",
      "primary_key_columns": ["order_id"],
      "scheduling_policy": "SNAPSHOT",
      "postgres_database": "mydb",
      "create_database_objects_if_missing": true
    }
  }'

SYNCED_TABLE_ID位置指定引数は、書式catalog.schema.tableを使用します。 Postgres では、テーブル {table}は、{schema}で設定したデータベース内のスキーマ postgres_databaseに作成されます (ここでは、mydb)。 このコマンドは、操作が既定で完了するまで待機します。 使用可能なすべてのオプションについては、 databricks postgres create-synced-table を参照してください。

Python SDK

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.postgres import (
    SyncedTable,
    SyncedTableSyncedTableSpec,
    SyncedTableSyncedTableSpecSyncedTableSchedulingPolicy,
)

w = WorkspaceClient()

synced_table = w.postgres.create_synced_table(
    synced_table=SyncedTable(spec=SyncedTableSyncedTableSpec(
        source_table_full_name="main.sales.orders",
        branch="projects/my-project/branches/production",
        primary_key_columns=["order_id"],
        scheduling_policy=SyncedTableSyncedTableSpecSyncedTableSchedulingPolicy.SNAPSHOT,
        postgres_database="mydb",
        create_database_objects_if_missing=True,
    )),
    synced_table_id="my-catalog.sales.orders",
).wait()

print(f"Synced table created: {synced_table.name}")

synced_table_idはcatalog.schema.table形式を使用し、Unity カタログ同期テーブル名になります。 Postgres では、テーブル {table}は、{schema}で設定したデータベース内のスキーマ postgres_databaseに作成されます (ここでは、mydb)。

Java SDK

import com.databricks.sdk.WorkspaceClient;
import com.databricks.sdk.service.postgres.*;
import java.util.List;

WorkspaceClient w = new WorkspaceClient();

SyncedTable syncedTable = w.postgres().createSyncedTable(
    new CreateSyncedTableRequest()
        .setSyncedTableId("my-catalog.sales.orders")
        .setSyncedTable(new SyncedTable()
            .setSpec(new SyncedTableSyncedTableSpec()
                .setSourceTableFullName("main.sales.orders")
                .setBranch("projects/my-project/branches/production")
                .setPrimaryKeyColumns(List.of("order_id"))
                .setSchedulingPolicy(SyncedTableSyncedTableSpecSyncedTableSchedulingPolicy.SNAPSHOT)
                .setPostgresDatabase("mydb")
                .setCreateDatabaseObjectsIfMissing(true))))
    .waitForCompletion();

System.out.println("Synced table created: " + syncedTable.getName());

curl

curl -X POST "https://your-workspace.cloud.databricks.com/api/2.0/postgres/synced_tables?synced_table_id=my-catalog.sales.orders" \
  -H "Authorization: Bearer ${DATABRICKS_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{
    "spec": {
      "source_table_full_name": "main.sales.orders",
      "branch": "projects/my-project/branches/production",
      "primary_key_columns": ["order_id"],
      "scheduling_policy": "SNAPSHOT",
      "postgres_database": "mydb",
      "create_database_objects_if_missing": true
    }
  }'

これにより、実行時間の長い操作が返されます。 返された name フィールドを done: trueまでポーリングします。 実行時間の長い操作を参照してください。 認証のセットアップについては、「 認証」を参照してください。

後続の同期をスケジュールまたはトリガーする

初期スナップショットは作成時に自動的に実行されます。 スナップショット モードとトリガーモードでは、後続の同期を明示的にトリガーする必要があります。 連続 モードは自己管理です。

データベース テーブル同期パイプライン タスク

Lakeflow ジョブの データベース テーブル同期パイプライン タスクは、同期されたテーブルのパイプラインをワークフロー ステップとして実行します。 テーブル更新トリガーまたはスケジュールを使用してジョブを構成します。

ソース テーブルの更新時にトリガーする

ソース Unity カタログ テーブルが更新されたときにジョブを起動します。 トリガー モードでは、新しい変更のみが増分的に適用され、継続的モードの常時オン コストなしでほぼリアルタイムの鮮度が得られます。

1. サイドバーで[ ワークフロー]をクリックします。
2. [ ジョブの作成 ] をクリックするか、既存のジョブを開きます。
3. [ タスク ] タブで、[ + 別のタスクの種類の追加] をクリックします。
4. [ インジェストと変換] で、[ データベース テーブル同期パイプライン] を選択します。
5. [ パイプライン ] フィールドで、同期されたテーブルに関連付けられているパイプラインを選択します。
6. [スケジュールとトリガー] で、[トリガーの追加] をクリックします。
7. トリガーの種類として [テーブルの更新 ] を選択します。
8. [ テーブル] で、監視するソース Unity カタログ テーブルを選択します。
9. [保存] をクリックします。

スケジュールに基づいてトリガー

固定間隔で同期を実行します。 通常、夜間または毎週の完全更新が最も効率的なパターンである スナップショット モードに適しています。

1. 上記の手順 1 から 5 に従って、 データベース テーブル同期パイプライン タスクをジョブに追加します。
2. [スケジュールとトリガー] で、[トリガーの追加] をクリックします。
3. トリガーの種類として [ スケジュール済 ] を選択します。
4. cron スケジュールとタイムゾーンを設定し、[ 保存] をクリックします。

同期の状態を確認する

同期されたテーブルの現在の状態と最後の同期時刻を確認するには:

UI

[カタログ] で、同期したテーブルに移動し、[概要] タブを選択します。現在の同期状態、パイプラインの状態、最後の同期タイムスタンプが表示されます。

Python SDK

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

table = w.postgres.get_synced_table("synced_tables/my-catalog.sales.orders")
print(f"State: {table.status.detailed_state}")
print(f"Last sync: {table.status.last_sync_time}")
print(f"Message: {table.status.message}")

Java SDK

import com.databricks.sdk.WorkspaceClient;
import com.databricks.sdk.service.postgres.SyncedTable;

WorkspaceClient w = new WorkspaceClient();

SyncedTable table = w.postgres().getSyncedTable("synced_tables/my-catalog.sales.orders");
System.out.println("State: " + table.getStatus().getDetailedState());
System.out.println("Last sync: " + table.getStatus().getLastSyncTime());
System.out.println("Message: " + table.getStatus().getMessage());

curl

curl "https://your-workspace.cloud.databricks.com/api/2.0/postgres/synced_tables/my-catalog.sales.orders" \
  -H "Authorization: Bearer ${DATABRICKS_TOKEN}"

データ型と互換性

Unity カタログのデータ型は、同期テーブルの作成時に Postgres 型にマップされます。 複合型 (ARRAY、MAP、STRUCT) は JSONB として Postgres に格納されます。

無効な文字を処理する

Null バイト (0x00) などの特定の文字は、Unity カタログの STRING、ARRAY、MAP、または STRUCT 列では使用できますが、Postgres TEXT または JSONB 列ではサポートされていません。 これにより、次のようなエラーで同期エラーが発生する可能性があります。

ERROR: invalid byte sequence for encoding "UTF8": 0x00
ERROR: unsupported Unicode escape sequence DETAIL: \u0000 cannot be converted to text

• 最初のエラーは、Postgres TEXTに直接マップされる最上位の文字列列に null バイトが表示されるときに発生します。
• 2 番目のエラーは、STRUCTとしてシリアル化される複合型 (ARRAY、MAP、またはJSONB) 内に入れ子になった文字列に null バイトが含まれている場合に発生します。 シリアル化中、すべての文字列は Postgres TEXTにキャストされます。ここで、 \u0000 は許可されません。

ソリューション

• 文字列フィールドをサニタイズする: 同期する前に、サポートされていない文字を削除します。 STRING 列の null バイトの場合:

  SELECT REPLACE(column_name, CAST(CHAR(0) AS STRING), '') AS cleaned_column FROM your_table
  

• BINARY への変換: 生バイトを保持する必要がある STRING 列の場合は、BINARY 型に変換します。

容量計画

同期テーブルの実装を計画するときは、次のリソース要件を考慮してください。

• 接続の使用状況: 同期された各テーブルは、インスタンスの接続制限にカウントされる最大 16 個の Lakebase データベースへの接続を使用します。
• サイズ クォータ: 同期されたすべてのテーブルの論理データの合計には、16 TB のクォータがあります。 より大きなクォータが必要な場合は、Databricks サポートにお問い合わせください。 個々のテーブルにはクォータはありませんが、Databricks では更新が必要なテーブルに対して 1 TB を超えないことをお勧めします。
• 完全更新サイズ: 完全更新をトリガーする場合、Postgres の古いバージョンは、新しい同期が完了するまで削除されません。 どちらのバージョンも、更新中に論理データベース サイズ クォータに一時的にカウントされます。
• ソースごとのテーブル: 1 つのソース テーブルに最大 20 個の同期テーブルを含めることができます。
• 名前付け要件: データベース、スキーマ、およびテーブル名には、英数字とアンダースコア ([A-Za-z0-9_]+) のみを含めることができます。
• ソース識別子のガイダンス: ソース Unity カタログ テーブルの列名またはテーブル名に大文字または特殊文字を使用しないでください。 それらを保持する場合は、Postgres で参照するときにそれらの識別子を引用符で囲む必要があります。
• スキーマの進化: トリガーモードと連続モードでは、追加スキーマの変更 (列の追加など) のみがサポートされます。
• 重複キー: ソース テーブル内の 2 つの行に同じ主キーがある場合、 時系列キーを使用して重複除去を構成しない限り、同期パイプラインは失敗します。
• APIの冪等性: 同期テーブル API は冪等であるため、一時的なエラーが発生した場合は再試行して、処理を適時に実行できるようにしてください。
• 更新率: Lakebase 自動スケーリングの場合、同期パイプラインでは、容量ユニット (CU) あたり約 150 行/秒の連続書き込みとトリガーされた書き込みがサポートされ、CU あたり 1 秒あたり最大 2,000 行のスナップショット書き込みがサポートされます。

Postgres で同期されたテーブルに対して許可される操作

Azure Databricksでは、誤って上書きやデータの不整合を防ぐために、同期されたテーブルに対して Postgres で次の操作のみを実行することをお勧めします。

• 読み取り専用クエリ
• インデックスの作成
• テーブルを削除する (Unity カタログから同期されたテーブルを削除した後に領域を解放するため)

Postgres で同期されたテーブルを他の方法で変更することは可能ですが、同期パイプラインに干渉します。

所有権とアクセス許可

同期されたテーブルは内部 databricks_writer_<dbid> ロールによって所有されます。これは、同期パイプラインによって管理されるためです ( Postgres ロールを参照)。 行レベルのセキュリティの構成など、所有者専用のコマンドは、同期されたテーブルで直接実行することはできません。

同期されたテーブルを作成するユーザーのアクセス権

同期されたテーブルを作成すると、そのテーブルを使用するためのアクセス権がAzure Databricks ID に自動的に付与されます。 databricks_superuserアクションは必要ありません。 ID には、同期されたテーブルに対する次の特権が付与されます。

INSERTまたはUPDATEは付与されません。 パイプラインはテーブルのデータを所有しているため、次回の更新時に直接書き込みが上書きされます。 DELETE と TRUNCATE は、テーブルをクリアするだけです。 次の更新では、ソースからテーブルが再作成されます。

このアクセスは、同期されたテーブルに対する Unity カタログのアクセス許可から派生し、Unity カタログで管理されます。 変更するには、ユーザーの Unity カタログのアクセス許可を更新します。 Postgres のAzure Databricks ID から直接REVOKEすることはできません。

同期されたテーブル アクセスの管理

同期されたテーブルが作成されると、 databricks_superuser は Postgres から同期されたテーブルを読み取ることができます。 databricks_superuserにはpg_read_all_dataがあり、このロールはすべてのテーブルから読み取られます。 また、このロールはすべてのテーブルに書き込むための pg_write_all_data 特権も持ちます。 つまり、 databricks_superuser は Postgres で同期されたテーブルに書き込むこともできます。 Lakebase では、ターゲット テーブルに緊急の変更を加える必要がある場合に備えて、この書き込み動作がサポートされています。 ただし、Azure Databricksでは、代わりにソース テーブルで修正を行うことをお勧めします。

• databricks_superuserは、次の権限を他のユーザーに付与することもできます。

  GRANT USAGE ON SCHEMA synced_table_schema TO user;
  

  GRANT SELECT ON synced_table_name TO user;
  

• databricks_superuserは、次の特権を取り消すことができます。

  REVOKE USAGE ON SCHEMA synced_table_schema FROM user;
  

  REVOKE {SELECT | INSERT | UPDATE | DELETE} ON synced_table_name FROM user;
  

同期されたテーブル操作を管理する

databricks_superuserでは、同期されたテーブルに対して特定の操作を実行する権限を持つユーザーを管理できます。 同期テーブルでサポートされる操作は次のとおりです。

• CREATE INDEX
• ALTER INDEX
• DROP INDEX
• DROP TABLE

他のすべての DDL 操作は、同期されたテーブルに対して拒否されます。

これらの権限を追加のユーザーに付与するには、 databricks_superuser が最初に databricks_auth で拡張機能を作成する必要があります。

CREATE EXTENSION IF NOT EXISTS databricks_auth;

その後、 databricks_superuser は、同期されたテーブルを管理するユーザーを追加できます。

SELECT databricks_synced_table_add_manager('"synced_table_schema"."synced_table"'::regclass, '[user]');

databricks_superuserは、同期されたテーブルの管理からユーザーを削除できます。

SELECT databricks_synced_table_remove_manager('[table]', '[user]');

databricks_superuserでは、すべてのマネージャーを表示できます。

SELECT * FROM databricks_synced_table_managers;

同期されたテーブルを削除する

Unity カタログから同期テーブルを削除すると、対応する Postgres テーブルも削除されます。

UI

[カタログ] で、同期したテーブルを見つけ、[をクリックして、[削除] を選択します。

Python SDK

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

w.postgres.delete_synced_table("synced_tables/my-catalog.sales.orders").wait()

Java SDK

import com.databricks.sdk.WorkspaceClient;

WorkspaceClient w = new WorkspaceClient();

w.postgres().deleteSyncedTable("synced_tables/my-catalog.sales.orders").waitForCompletion();

curl

curl -X DELETE "https://your-workspace.cloud.databricks.com/api/2.0/postgres/synced_tables/my-catalog.sales.orders" \
  -H "Authorization: Bearer ${DATABRICKS_TOKEN}"

詳細情報

カタログ統合

• カタログの重複: 別のデータベース カタログとしても登録されている Postgres データベースをターゲットとする標準カタログに同期テーブルを作成すると、同期されたテーブルが標準カタログとデータベース カタログの両方の下に Unity カタログに表示されます。

その他のオプション

Databricks 以外のシステムにデータを同期する場合は、Census や Hightouch などの パートナー接続の逆 ETL ソリューション を参照してください。