0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@manabian

Databricks にて LTAP Direct Writes による Lakebase への大規模データ同期を検証

0
Last updated at Posted at 2026-10-26

概要

Databricks の Lakehouse テーブルから Lakebase テーブルへ効率的にデータを同期するための機能「LTAP Direct Writes」について、実際に検証した結果を紹介します。

LTAP Direct Writes は、Unity Catalog で管理されるデータを Lakebase（PostgreSQL）へ高速に同期するための機能です。従来の Synced Tables では PostgreSQL のコンピュートエンドポイントを経由してデータを書き込んでいましたが、本機能ではストレージ層へ直接書き込むことで、大規模テーブルのロード時間を短縮し、同期処理がクエリ性能に与える影響を軽減します。

Snapshot モードでは初期ロードと全件更新が高速化の対象となり、Triggered モードおよび Continuous モードでは初期ロードが高速化の対象となります。

本記事では、最大 10 億レコードのデータを使用し、データ量やセカンダリーインデックスの有無による同期処理の動作を検証します。

事前準備

Unity Catalog にカタログとスキーマを作成

同期元のテーブルを管理するため、Unity Catalog にカタログとスキーマを作成します。

%sql
CREATE CATALOG IF NOT EXISTS ltap_direct_01;
CREATE SCHEMA IF NOT EXISTS ltap_direct_01.schema_01;

続いて、同期先の Lakebase テーブルを管理するためのスキーマを作成します。

%sql
CREATE SCHEMA IF NOT EXISTS ltap_direct_01.lakebase_01;

image.png

Lakebase への同期元テーブルを作成

Lakebase への同期元となる Delta テーブルを作成します。顧客 ID、顧客名、地域、顧客ランクなどの列に加え、データ量を調整するためのペイロード列を定義しています。

また、変更データを追跡できるように、Change Data Feed（CDF）を有効化しています。

%sql
CREATE TABLE ltap_direct_01.schema_01.customers_large (
  customer_id   BIGINT NOT NULL,
  customer_name STRING,
  region        STRING,
  customer_rank STRING,
  total_amount  DECIMAL(18, 2),
  is_active     BOOLEAN,
  updated_at    TIMESTAMP,
  payload       STRING
)
USING DELTA
TBLPROPERTIES (
  'delta.enableChangeDataFeed' = 'true'
);

image.png

Lakebase のプロジェクトを作成

同期先となる Lakebase のプロジェクトを作成します。以降の検証では、このプロジェクト内に作成する PostgreSQL テーブルへデータを同期します。

image.png

検証

LTAP Direct Writes による初回同期

1,000 万レコードのデータを作成

初回同期を検証するため、同期元テーブルに 1,000 万レコードのデータを書き込みます。

各レコードには、顧客情報に加えて、SHA-256 のハッシュ値を組み合わせた 256 文字のペイロードを設定しています。

%sql
-- 初回実行
INSERT INTO ltap_direct_01.schema_01.customers_large
SELECT
  id + 1 AS customer_id,

  concat(
    'customer_',
    lpad(cast(id + 1 AS STRING), 10, '0')
  ) AS customer_name,

  CASE pmod(id, 5)
    WHEN 0 THEN 'tokyo'
    WHEN 1 THEN 'osaka'
    WHEN 2 THEN 'nagoya'
    WHEN 3 THEN 'fukuoka'
    ELSE 'sapporo'
  END AS region,

  CASE pmod(id, 3)
    WHEN 0 THEN 'gold'
    WHEN 1 THEN 'silver'
    ELSE 'bronze'
  END AS customer_rank,

  cast(
    pmod(id, 100000) / 100.0
    AS DECIMAL(18, 2)
  ) AS total_amount,

  pmod(id, 10) <> 0 AS is_active,

  current_timestamp() AS updated_at,

  -- 行ごとに異なる256文字のペイロード
  concat(
    sha2(concat('a:', cast(id AS STRING)), 256),
    sha2(concat('b:', cast(id AS STRING)), 256),
    sha2(concat('c:', cast(id AS STRING)), 256),
    sha2(concat('d:', cast(id AS STRING)), 256)
  ) AS payload

FROM range(0, 10000000, 1, 128);

image.png

同期テーブルを作成

Unity Catalog から、作成した同期元テーブルを選択します。

image.png

画面右上の 作成 から 同期テーブル を選択します。

image.png

同期先として lakebase_01 スキーマを指定し、同期元と同名のテーブルを設定します。また、LTAPダイレクト書き込み を有効にしたうえで、作成 を選択します。

image.png

同期テーブルの作成後、データ取り込み セクションから同期処理が開始されていることを確認できます。

image.png

データ取り込み セクションに表示されるパイプライン ID を選択すると、同期処理を実行しているパイプラインの詳細画面に移動できます。

image.png

初回同期の実行結果

今回の検証環境では、1,000 万レコードの初回同期が 4 分 28 秒で完了しました。実行時間の内訳を確認すると、初期化処理が大半を占めており、データの書き込み以外に要する時間が比較的大きいことが分かります。

image.png

同期完了後、Lakebase 側でクエリを実行したところ、同期元のデータが正常に反映されていることを確認できました。

image.png

1 億レコード（約 12 GB）の再同期

続いて、データ量が増加した場合の同期処理を確認します。

同期元テーブルを 1 億レコード（約 12 GB）のデータで上書きし、Lakebase へ再同期します。

%sql
-- 全件書き換え
INSERT OVERWRITE ltap_direct_01.schema_01.customers_large
SELECT
  id + 1 AS customer_id,

  concat(
    'customer_',
    lpad(cast(id + 1 AS STRING), 10, '0')
  ) AS customer_name,

  CASE pmod(id, 5)
    WHEN 0 THEN 'tokyo'
    WHEN 1 THEN 'osaka'
    WHEN 2 THEN 'nagoya'
    WHEN 3 THEN 'fukuoka'
    ELSE 'sapporo'
  END AS region,

  CASE pmod(id, 3)
    WHEN 0 THEN 'gold'
    WHEN 1 THEN 'silver'
    ELSE 'bronze'
  END AS customer_rank,

  cast(
    pmod(id, 100000) / 100.0
    AS DECIMAL(18, 2)
  ) AS total_amount,

  pmod(id, 10) <> 0 AS is_active,

  current_timestamp() AS updated_at,

  -- 行ごとに異なる256文字のペイロード
  concat(
    sha2(concat('a:', cast(id AS STRING)), 256),
    sha2(concat('b:', cast(id AS STRING)), 256),
    sha2(concat('c:', cast(id AS STRING)), 256),
    sha2(concat('d:', cast(id AS STRING)), 256)
  ) AS payload

FROM range(0, 100000000, 1, 128);

image.png

データの上書きが完了したら、同期先テーブルの画面から 今すぐ同期 を選択し、再同期を実行します。

image.png

再同期の実行結果は以下のとおりです。

image.png

セカンダリーインデックスを追加した場合の再同期

次に、Lakebase 側にセカンダリーインデックスが存在する場合の同期処理を検証します。

まず、同期元テーブルのデータを削除し、今すぐ同期 を実行して同期先にも変更を反映します。

%sql
TRUNCATE TABLE ltap_direct_01.schema_01.customers_large;

image.png

続いて、Lakebase 側で region 列にセカンダリーインデックスを作成します。

-- セカンダリーインデックス
CREATE INDEX idx_customers_region
    ON lakebase_01.customers_large (region);

image.png

インデックスの作成後、Lakebase の Tables タブから、対象テーブルにセカンダリーインデックスが追加されていることを確認できます。

image.png

インデックスを追加した状態で、同期元テーブルに再び 1 億レコードのデータを書き込みます。

%sql
-- 全件書き換え
INSERT OVERWRITE ltap_direct_01.schema_01.customers_large
SELECT
  id + 1 AS customer_id,

  concat(
    'customer_',
    lpad(cast(id + 1 AS STRING), 10, '0')
  ) AS customer_name,

  CASE pmod(id, 5)
    WHEN 0 THEN 'tokyo'
    WHEN 1 THEN 'osaka'
    WHEN 2 THEN 'nagoya'
    WHEN 3 THEN 'fukuoka'
    ELSE 'sapporo'
  END AS region,

  CASE pmod(id, 3)
    WHEN 0 THEN 'gold'
    WHEN 1 THEN 'silver'
    ELSE 'bronze'
  END AS customer_rank,

  cast(
    pmod(id, 100000) / 100.0
    AS DECIMAL(18, 2)
  ) AS total_amount,

  pmod(id, 10) <> 0 AS is_active,

  current_timestamp() AS updated_at,

  -- 行ごとに異なる256文字のペイロード
  concat(
    sha2(concat('a:', cast(id AS STRING)), 256),
    sha2(concat('b:', cast(id AS STRING)), 256),
    sha2(concat('c:', cast(id AS STRING)), 256),
    sha2(concat('d:', cast(id AS STRING)), 256)
  ) AS payload

FROM range(0, 100000000, 1, 128);

image.png

同期元テーブルへの書き込みが完了したら、同期先テーブルの 今すぐ同期 を選択して再同期を実行します。

image.png

セカンダリーインデックスを追加した状態での同期結果は以下のとおりです。

image.png

セカンダリーインデックスを設定した状態で 10 億レコード（約 120 GB）を再同期

最後に、セカンダリーインデックスを設定した状態で、さらにデータ量を増やして検証します。

同期元テーブルを 10 億レコード（約 120 GB）のデータで上書きし、大規模データの再同期を実行します。

%sql
-- 全件書き換え
INSERT OVERWRITE ltap_direct_01.schema_01.customers_large
SELECT
  id + 1 AS customer_id,

  concat(
    'customer_',
    lpad(cast(id + 1 AS STRING), 10, '0')
  ) AS customer_name,

  CASE pmod(id, 5)
    WHEN 0 THEN 'tokyo'
    WHEN 1 THEN 'osaka'
    WHEN 2 THEN 'nagoya'
    WHEN 3 THEN 'fukuoka'
    ELSE 'sapporo'
  END AS region,

  CASE pmod(id, 3)
    WHEN 0 THEN 'gold'
    WHEN 1 THEN 'silver'
    ELSE 'bronze'
  END AS customer_rank,

  cast(
    pmod(id, 100000) / 100.0
    AS DECIMAL(18, 2)
  ) AS total_amount,

  pmod(id, 10) <> 0 AS is_active,

  current_timestamp() AS updated_at,

  -- 行ごとに異なる256文字のペイロード
  concat(
    sha2(concat('a:', cast(id AS STRING)), 256),
    sha2(concat('b:', cast(id AS STRING)), 256),
    sha2(concat('c:', cast(id AS STRING)), 256),
    sha2(concat('d:', cast(id AS STRING)), 256)
  ) AS payload

FROM range(0, 1000000000, 1, 128);

image.png

データの上書き後、同期先テーブルから 今すぐ同期 を選択し、10 億レコードの再同期を実行します。

image.png

10 億レコードの再同期を実行した結果は以下のとおりです。

image.png

まとめ

LTAP Direct Writes を使用し、Lakehouse から Lakebase へのデータ同期を検証しました。

今回の検証では、1,000 万レコードの初回同期に加え、1 億レコードへのデータ量の拡大、セカンダリーインデックスの追加、さらに 10 億レコードへの拡大という条件で同期処理を実行しました。

LTAP Direct Writes は、Lakehouse のデータを Lakebase に連携し、PostgreSQL を利用するアプリケーションから活用する際に有用な機能です。特に、大規模な初期ロードや全件更新を伴うワークロードでは、活用を検討する価値があると考えられます。

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?