Google Cloud の Lakehouse for Apache Iceberg を Azure Databricks へ増分 DEEP CLONE する

Last updated at 2026-07-27Posted at 2026-07-27

はじめに

Databricks では、Apache Iceberg テーブルを Delta テーブルへ DEEP CLONE し、2回目以降は差分ファイルだけを取り込む増分同期が可能です。

本記事では、Google Cloud の Lakehouse Runtime Catalog で管理される Lakehouse for Apache Iceberg テーブルを、Azure Databricks の Delta テーブルへ DEEP CLONE する方法を検証します。

次の内容を順に確認します。

Google CloudでのLakehouse for Apache Icebergテーブルの準備
Azure DatabricksからCloud Storageへ接続するためのクラシッククラスター設定
最新のIcebergメタデータファイルを指定した DEEP CLONE
Iceberg REST APIによる最新の metadata-location の取得
BigQueryで更新したデータの増分取り込み
DESCRIBE HISTORY を用いた増分コピーの確認
クローン先で OPTIMIZE を実行した場合の注意点

Google Cloud にて Lakehouse for Apache Iceberg テーブルを作成

以下の記事を参照して、Lakehouse for Apache Iceberg テーブルを作成します。

Azure Databricks でクラシッククラスターの準備

以下の記事を参照して、クラシッククラスターに必要な Spark構成を設定します。

Azure Databricks で DEEP CLONE

Lakehouse for Apache Iceberg テーブルの確認

CLONE 元となる Lakehouse for Apache Iceberg テーブルのスナップショットを確認しておきます。

また、最新の metadata パスを確認します。

DEEP CLONE の実行

Iceberg の最新の metadata パスを指定することで CLONE が可能です。

%sql
CREATE OR REPLACE TABLE <catalog>.<schema>.<table> CLONE iceberg.`<Cloud Storage URI>`;

データも Lakehouse for Apache Iceberg テーブルと一致することを確認できました。

%sql
SELECT * FROM <catalog>.<schema>.<table>;

フォルダ指定した場合の挙動

metadata フォルダや data フォルダを指定しても CLONE はできませんでした。

Iceberg REST API および PyIceberg で metadata のファイルのパスを取得する方法

運用する上では、最新の metadata のファイルのパスを UI から取得するのは厳しいため、Iceberg REST API および PyIceberg で取得する方法を紹介します。

Spark Conf に設定されているGoogle Cloudサービスアカウント情報を使って、OAuth 2.0のアクセストークンを発行する処理

from google.oauth2 import service_account
from google.auth.transport.requests import Request

# Spark Conf からサービスアカウント情報を取得
service_account_email = spark.conf.get(
    "spark.hadoop.fs.gs.auth.service.account.email"
)

private_key = spark.conf.get(
    "spark.hadoop.fs.gs.auth.service.account.private.key"
)

private_key_id = spark.conf.get(
    "spark.hadoop.fs.gs.auth.service.account.private.key.id"
)

service_account_info = {
    "type": "service_account",
    "project_id": "data",
    "private_key_id": private_key_id,
    "private_key": private_key.replace("\\n", "\n"),
    "client_email": service_account_email,
    "token_uri": "https://oauth2.googleapis.com/token",
}

# Google認証用Credentialsを作成
credentials = service_account.Credentials.from_service_account_info(
    service_account_info,
    scopes=["https://www.googleapis.com/auth/cloud-platform"],
)

# Google OAuthエンドポイントへトークン要求
credentials.refresh(Request())

# 短時間有効なアクセストークンを取得
access_token = credentials.token

上記で発行したアクセストークンを使用して、Iceberg REST API で metadata のファイルの最新パスを取得する処理

import requests

project_id = "<project_id>"
catalog_id = "<catalog_id>"
namespace = "<namespace>"
table_name = "<table_name>"

url = (
    "https://biglake.googleapis.com/iceberg/v1/restcatalog/v1/"
    f"projects/{project_id}/catalogs/{catalog_id}/"
    f"namespaces/{namespace}/tables/{table_name}"
)

response = requests.get(
    url,
    headers={
        "Authorization": f"Bearer {access_token}",
        "x-goog-user-project": project_id,
        # 認証情報ベンディング モードの場合、必要な設定
        "X-Iceberg-Access-Delegation": "vended-credentials",
        "Accept": "application/json",
    },
    timeout=30,
)

print(response.status_code)
print(response.text)

response.raise_for_status()

result = response.json()
metadata_location = result.get("metadata-location")

print(metadata_location)

上記で発行したアクセストークンを使用して、PyIceberg で metadata のファイルの最新パスを取得する処理

from pyiceberg.catalog import load_catalog

project_id = "<project_id>"
catalog_id = "<catalog_id>"
namespace = "<namespace>"
table_name = "<table_name>"

rest_uri = "https://biglake.googleapis.com/iceberg/v1/restcatalog"

warehouse = (
    f"bl://projects/{project_id}/catalogs/{catalog_id}"
)

catalog = load_catalog(
    "google_lakehouse", # 任意の名称
    **{
        "type": "rest",
        "uri": rest_uri,
        "warehouse": warehouse,
        "token": access_token,
        "header.x-goog-user-project": project_id,
        "header.X-Iceberg-Access-Delegation": "vended-credentials",
    },
)

table = catalog.load_table((namespace, table_name))

print("Metadata location:")
print(table.metadata_location)

増分更新の確認

BigQuery にて Lakehouse for Apache Iceberg テーブルに行を追加します。

INSERT INTO 
  `PROJECT_NAME.CATALOG_ID.NAMESPACE_OR_SCHEMA_NAME.TABLE_NAME`
VALUES
  (3, '3rd row', current_timestamp())
;

Databricks で DEEP CLONE 実行後、テーブル履歴を確認します。

%sql
SELECT
  version,
  timestamp,
  operation,
  operationParameters,
  operationMetrics
FROM (
  DESCRIBE HISTORY
    <catalog>.<schema>.<table>
)
ORDER BY version DESC;