同期されたテーブルを使用して Lakehouse データを提供する (Lakebase プロビジョニング済み)

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

同期テーブルとは

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

同期パイプラインでは、マネージド Lakeflow Spark 宣言パイプラインを使用して、Unity カタログ同期テーブルと Postgres テーブルの両方をソース テーブルからの変更で継続的に更新します。 作成後、同期されたテーブルは Postgres ツールを使用して直接照会できます。

同期テーブルの主な特性は次のとおりです。

• テーブルは Postgres で読み取り専用として扱われ、ソースとのデータ整合性を維持します
• マネージド Lakeflow Spark 宣言パイプラインを使用して自動的に同期する
• 標準の PostgreSQL インターフェイスを介してクエリ可能
• ガバナンスとライフサイクル管理のために Unity カタログを介して管理

開始する前に

• 任意のカタログに Unity カタログ テーブルがあります。
• データベース インスタンスに対する CAN USE 権限があります。

同期テーブルを作成する

UI

Unity カタログ テーブルを Postgres に同期するには、次の操作を行います。

1. ワークスペースのサイドバーで [ カタログ ] をクリックします。

2. 同期テーブルを作成する Unity カタログ テーブルを見つけて選択します。

3. [ 作成>同期テーブル] をクリックします。

4. カタログ、スキーマを選択し、新しい同期テーブルのテーブル名を入力します。

   • 同期テーブルは、いくつかの追加構成を使用して Standard カタログに作成することもできます。 標準カタログ、スキーマを選択し、新しく作成した同期テーブルのテーブル名を入力します。

5. データベース インスタンスを選択し、同期テーブルを作成する Postgres データベースの名前を入力します。 Postgres データベース フィールドの既定値は、現在選択されているターゲット カタログです。 この名前の下に Postgres データベースが存在しない場合、Azure Databricks によって新しいデータベースが作成されます。

6. 主キーを選択します。 主キーは、読み取り、更新、および削除の行への効率的なアクセスを可能にするため、 必要 です。

   Important

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

7. ソース テーブル内の 2 つの行に同じ主キーがある場合は、 Timeseries キー を選択して重複除去を構成します。 時系列キーを指定すると、同期されたテーブルには、各主キーの最新の時系列キー値を持つ行のみが含まれます。

8. [スナップショット]、[トリガー済み]、[継続的] から同期モードを選択します。 各同期モードの詳細については、説明 されている同期モードを参照してください。

9. この同期テーブルを新規または既存のパイプラインから作成するかどうかを選択します。

   • 新しいパイプラインを作成し、マネージド カタログを使用する場合は、ステージング テーブルのストレージの場所を選択します。 標準カタログを使用している場合、ステージング テーブルはカタログに自動的に格納されます。
   • 既存のパイプラインを使用している場合は、新しい同期モードがパイプライン モードと一致することを確認します。

10. (省略可能) サーバーレス予算ポリシーを選択します。 サーバーレス予算ポリシーを作成するには、「サーバーレス予算ポリシー を使用した属性の使用」を参照してください。 これにより、課金の使用状況を特定の使用状況ポリシーに属性付けできます。

    • 同期テーブルの場合、課金対象エンティティは基になる Lakeflow Spark 宣言型パイプラインです。 予算ポリシーを変更するには、基になるパイプライン オブジェクトを変更します。 サーバーレス パイプラインの構成を参照してください。

11. 同期されたテーブルの状態がオンラインになった後、データベース インスタンスにログインし、新しく作成されたテーブルに対してクエリを実行します。 SQL エディター、外部ツール、またはノートブックを使用してテーブルのクエリを実行します。

Python SDK

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.database import SyncedDatabaseTable, SyncedTableSpec, NewPipelineSpec, SyncedTableSchedulingPolicy

# Initialize the Workspace client
w = WorkspaceClient()

# Create a synced table in a database catalog
synced_table = w.database.create_synced_database_table(
    SyncedDatabaseTable(
        name="database_catalog.schema.synced_table",  # Full three-part name
        spec=SyncedTableSpec(
            source_table_full_name="source_catalog.source_schema.source_table",
            primary_key_columns=["id"],  # Primary key columns
            scheduling_policy=SyncedTableSchedulingPolicy.TRIGGERED,  # SNAPSHOT, TRIGGERED, or CONTINUOUS
            # Optional: timeseries_key="timestamp"  # For deduplication
            new_pipeline_spec=NewPipelineSpec(
                storage_catalog="storage_catalog",
                storage_schema="storage_schema"
            )
        ),
    )
)
print(f"Created synced table: {synced_table.name}")

# Create a synced table in a standard UC catalog
synced_table = w.database.create_synced_database_table(
    SyncedDatabaseTable(
        name="standard_catalog.schema.synced_table",  # Full three-part name
        database_instance_name="my-database-instance",  # Required for standard catalogs
        logical_database_name="postgres_database",  # Required for standard catalogs
        spec=SyncedTableSpec(
            source_table_full_name="source_catalog.source_schema.source_table",
            primary_key_columns=["id"],
            scheduling_policy=SyncedTableSchedulingPolicy.CONTINUOUS,
            create_database_objects_if_missing=True,  # Create database/schema if needed
            new_pipeline_spec=NewPipelineSpec(
                storage_catalog="storage_catalog",
                storage_schema="storage_schema"
            )
        ),
    )
)
print(f"Created synced table: {synced_table.name}")

# Check the status of a synced table
synced_table_name = "database_catalog.schema.synced_table"
status = w.database.get_synced_database_table(name=synced_table_name)
print(f"Synced table status: {status.data_synchronization_status.detailed_state}")
print(f"Status message: {status.data_synchronization_status.message}")

CLI

# Create a synced table in a database catalog
databricks database create-synced-database-table  \
  --json '{
    "spec": {
      "name": "database_catalog.schema.synced_table",
      "source_table_full_name": "source_catalog.source_schema.source_table",
      "primary_key_columns": ["id"],
      "scheduling_policy": "TRIGGERED"
    },
    "new_pipeline_spec": {
      "storage_catalog": "storage_catalog",
      "storage_schema": "storage_schema"
    }
  }'

# Create a synced table in a standard UC catalog
# new_pipeline_spec, storage_catalog, and storage_schema are optional
databricks database create-synced-database-table \
  --database-instance-name "my-database-instance" \
  --logical-database-name "databricks_postgres" \
  --json '{
    "name": "standard_catalog.schema.synced_table",
    "spec": {
      "source_table_full_name": "source_catalog.source_schema.source_table",
      "primary_key_columns": ["id"],
      "scheduling_policy": "CONTINUOUS",
      "create_database_objects_if_missing": true
    }
  }'

# Check the status of a synced table
databricks database get-synced-database-table "database_catalog.schema.synced_table"

curl

データベース カタログに同期テーブルを作成します。

export CATALOG_NAME=<Database catalog>
export SRC_TBL="source_catalog.source_schema.source_table"
export DEST_TBL="$CATALOG_NAME.some_schema.synced_table"
export PKS='["id"]'
export ST_CATALOG="storage_catalog"
export ST_SCHEMA="storage_schema"

curl -X POST https://$WORKSPACE/api/2.0/database/synced_tables \
-H "Content-Type: text/json" \
-H "Authorization: Bearer ${DATABRICKS_TOKEN}" \
--data-binary @- << EOF
{
  "name": "$DEST_TBL",
  "spec": {
    "source_table_full_name": "$SRC_TBL",
    "primary_key_columns": $PKS,
    "scheduling_policy": "TRIGGERED",
  },
  "new_pipeline_spec": {
    "storage_catalog": "$ST_CATALOG",
    "storage_schema": "$ST_SCHEMA",
  }
}
EOF

標準の Unity カタログ カタログに同期テーブルを作成します。

export CATALOG_NAME=<Standard catalog>
export DATABASE_INSTANCE=<database instance>
export POSTGRES_DATABASE=$CATALOG_NAME
export SRC_TBL="source_catalog.source_schema.source_table"
export DEST_TBL="$CATALOG_NAME.some_schema.sync_table"
export PKS='["id"]'
export ST_CATALOG="storage_catalog"
export ST_SCHEMA="storage_schema"

curl -X POST https://$WORKSPACE/api/2.0/database/synced_tables \
-H "Content-Type: text/json" \
-H "Authorization: Bearer ${DATABRICKS_TOKEN}" \
--data-binary @- << EOF
{
  "name": "$DEST_TBL",
  "database_instance_name": "$DATABASE_INSTANCE",
  "logical_database_name": "$POSTGRES_DATABASE",
  "spec": {
    "source_table_full_name": "$SRC_TBL",
    "primary_key_columns": $PKS,
    "scheduling_policy": "TRIGGERED",
  },
  "new_pipeline_spec": {
    "storage_catalog": "$ST_CATALOG",
    "storage_schema": "$ST_SCHEMA",
  }
}
EOF

同期されたテーブルの状態を確認します。

export SYNCEDTABLE='pg_db.silver.sbtest1_online'

curl --request GET \
  "https://e2-dogfood.staging.cloud.databricks.com/api/2.0/database/synced_tables/$SYNCEDTABLE" \
  --header "Authorization: Bearer dapi..."

同期モードの説明

同期テーブルは、次のいずれかの同期モードで作成できます。これは、Postgres でソースから同期されたテーブルへのデータの同期方法を決定します。

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

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

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

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

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

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

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

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

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

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

SDK を使用する

たとえば、アップストリームのノートブックやパイプラインの末尾で、プログラムによって同期の実行をトリガーします。

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

# Get the pipeline ID from the synced table
table = w.database.get_synced_database_table(name="catalog.schema.synced_table")
pipeline_id = table.data_synchronization_status.pipeline_id

# Trigger a sync run
w.pipelines.start_update(pipeline_id=pipeline_id)

サポートされている操作

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

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

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

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

同期されたテーブルを削除するには、Unity カタログから削除し、データベース インスタンス内のテーブルを削除する必要があります。 Unity カタログから同期されたテーブルを削除すると、テーブルの登録が解除され、データの更新が停止されます。 ただし、テーブルは基になる Postgres データベースに残ります。 データベース インスタンスの領域を解放するには、インスタンスに接続し、 DROP TABLE コマンドを使用します。

UI

1. ワークスペースのサイドバーで [ カタログ ] をクリックします。
2. 削除する同期テーブルを見つけて選択します。
3. クリックします。>削除します。
4. psql、SQL エディター、またはノートブックからインスタンスに接続します。
5. PostgreSQL を使用してテーブルを削除します。

   DROP TABLE synced_table_database.synced_table_schema.synced_table
   

Python SDK

from databricks.sdk import WorkspaceClient

# Initialize the Workspace client
w = WorkspaceClient()

# Delete a synced table from UC
synced_table_name = "catalog.schema.synced_table"
w.database.delete_synced_database_table(name=synced_table_name)
print(f"Deleted synced table from UC: {synced_table_name}")

# To free up space in your database instance, you need to connect to the
# instance and drop the table using PostgreSQL:
#
# DROP TABLE synced_table_database.synced_table_schema.synced_table;

CLI

# Delete a synced table from UC
databricks database delete-synced-database-table "catalog.schema.synced_table"

# To free up space in your database instance, you need to connect to the
# instance and drop the table using PostgreSQL:
#
# DROP TABLE synced_table_database.synced_table_schema.synced_table;

curl

# Delete a synced table from UC
curl -X DELETE --header "Authorization: Bearer ${DATABRICKS_TOKEN}" \
  https://$WORKSPACE/api/2.0/database/synced_tables/$SYNCED_TABLE_NAME

# To free up space in your database instance, you need to connect to the
# instance and drop the table using PostgreSQL:
#
# DROP TABLE synced_table_database.synced_table_schema.synced_table;

所有権とアクセス許可

新しい Postgres データベース、スキーマ、またはテーブルを作成する場合、Postgres の所有権は次のように設定されます。

• Azure Databricks ログインが Postgres のロールとして存在する場合、データベース、スキーマ、またはテーブルを作成するユーザーに所有権が割り当てられます。 Postgres で Azure Databricks ID ロールを追加するには、「 PostgreSQL ロールの管理」を参照してください。
• それ以外の場合、所有権は Postgres の親オブジェクトの所有者 (通常は databricks_superuser) に割り当てられます。

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

同期されたテーブルが作成されると、 databricks_superuser は Postgres から同期されたテーブルを READ できます。 databricks_superuserにはpg_read_all_dataがあり、このロールはすべてのテーブルから読み取られます。 また、このロールはすべてのテーブルに書き込むための pg_write_all_data 特権も持ちます。 つまり、 databricks_superuser は Postgres で同期されたテーブルに書き込むこともできます。 Lakebase では、ターゲット テーブルに緊急の変更を加える必要がある場合に備えて、この書き込み動作がサポートされています。 ただし、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 の同期先テーブルにマップする方法を定義します。

無効な文字を処理する

null バイト (0x00) などの特定の文字は、デルタ テーブルの STRING、 ARRAY、 MAP、または STRUCT 列で使用できますが、Postgres TEXT または JSONB 列ではサポートされていません。 その結果、DeltaからPostgresにデータを同期することによって、挿入エラーなどの問題が発生する可能性があります。

org.postgresql.util.PSQLException: ERROR: invalid byte sequence for encoding "UTF8": 0x00

org.postgresql.util.PSQLException: ERROR: unsupported Unicode escape sequence DETAIL: \u0000 cannot be converted to text.

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

解決方法:

この問題には、次のいずれかの方法で対処できます。

• 文字列フィールドをサニタイズする

  Postgres に同期する前に、複合型内の文字列フィールドを含むすべての文字列フィールドからサポートされていない文字を削除または置換します。

  最上位の STRING 列から null バイトを削除するには、 REPLACE 関数を使用します。

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

• バイナリへの変換 ( STRING 列のみ)

  生のバイト コンテンツを保持する必要がある場合は、影響を受ける STRING 列を BINARYに変換します。

制限事項と考慮事項

サポートされているソース テーブル

同期テーブルの同期モードに応じて、さまざまな種類のソース テーブルがサポートされます。

• スナップショット モードの場合、ソース テーブルは SELECT *をサポートする必要があります。 たとえば、Delta テーブル、Iceberg テーブル、ビュー、具体化されたビュー、その他の同様の種類があります。

• トリガーモードまたは継続的同期モードでは、ソース テーブルで変更データ フィードも有効になっている必要があります。

名前付けと識別子の制限事項

• 使用できる文字: 同期されたテーブルの Postgres データベース、スキーマ、およびテーブル名には、英数字とアンダースコア ([A-Za-z0-9_]+) のみを含めることができます。 ハイフン (-) およびその他の特殊文字はサポートされていません。
• 列とテーブルの識別子: ソース Unity カタログ テーブルの列名またはテーブル名に大文字または特殊文字を使用しないでください。 保持されている場合は、Postgres で参照するときにこれらの識別子を引用符で囲む必要があります。

パフォーマンスと同期

• 同期速度: 同期されたテーブルへのデータの同期は、追加の処理により、ネイティブ PostgreSQL クライアントを使用してデータベース インスタンスに直接データを書き込むよりも時間がかかる場合があります。 継続的同期モードでは、Unity カタログのマネージド テーブルから同期されたテーブルに、15 秒以上の間隔でデータが更新されます。
• 接続の使用法: 各テーブル同期では、データベース インスタンスへの最大 16 個の接続を使用できます。これは、インスタンスの接続制限にカウントされます。
• API のべき等性: 同期テーブルに関する API にはべき等性があり、操作をタイムリーに実行するために、エラー発生時に再試行が必要になる場合があります。
• スキーマの変更:TriggeredモードまたはContinuous モードの同期テーブルの場合、Unity カタログ テーブルからの追加スキーマ変更 (新しい列の追加など) のみが同期テーブルに反映されます。
• キーの重複: ソース テーブル内の 2 つの行に同じ主キーがある場合、時系列キーを使用して重複除去を構成しない限り、同期パイプラインは失敗します。 ただし、時系列キーを使用すると、パフォーマンスが低下します。
• 更新率: Lakebase Provisioned の場合、同期パイプラインでは、容量ユニット (CU) あたり約 1,200 行/秒の継続的書き込みとトリガーされた書き込みがサポートされ、CU あたり 1 秒あたり最大 15,000 行のスナップショット書き込みがサポートされます。

容量と制限

• テーブルの制限:
  • ソース テーブルあたり 20 個の同期テーブルの制限。
  • 各テーブル同期では、最大 16 個のデータベース接続を使用できます。
• サイズの制限と完全な更新:
  • 同期されたテーブルを完全に更新した場合、Postgres の古いバージョンは、新しいテーブルが同期されるまで削除されません。 どちらのバージョンも、更新中に論理データベース サイズの制限に一時的にカウントされます。
  • 個々の同期テーブルには制限はありませんが、インスタンス内のすべてのテーブルの論理データ サイズの合計制限は 2 TB です。 ただし、完全なテーブル再作成ではなく更新が必要な場合は、Databricks では 1 TB を超えないことをお勧めします。
  • Unity カタログ テーブルの圧縮されていない行形式のサイズがデータベース インスタンスのサイズ制限 (2 TB) を超えた場合、同期は失敗します。 インスタンスにさらに書き込む前に、同期されたテーブルを Postgres で削除する必要があります。

カタログ統合

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