Google 会使用 AI 技术将内容翻译成您偏好的语言。AI 翻译可能包含错误。

基于开放式无边界湖仓一体构建跨云属性图

以下教程将向您展示如何构建一个 BigQuery 图，该图使用开放式无边界湖仓一体和 Apache Iceberg REST 目录端点来统一两个不同云中的数据孤岛，而无需移动数据。

准备工作

在开始之前，请设置您的环境并启用所需的 API。

设置项目和区域，并启用 API：

export PROJECT_ID="your-gcp-project-id"
export REGION="us-east4"

gcloud config set project "$PROJECT_ID"

gcloud services enable \
  biglake.googleapis.com \
  bigquery.googleapis.com \
  secretmanager.googleapis.com \
  storage.googleapis.com

为加载程序创建 Python 虚拟环境：

python3 -m venv iceberg-venv
source iceberg-venv/bin/activate
pip install --quiet "pyiceberg[pyarrow]"

创建 Google Cloud spoke

设置由 Cloud Storage 存储桶支持的开放式 Apache Iceberg REST 目录，并将三个 Iceberg 表加载到该目录中。

创建存储桶和目录：

export GCS_BUCKET="gs://${PROJECT_ID}-xcloud-lake"
export GCS_CATALOG="gcs_lake"

gcloud storage buckets create "$GCS_BUCKET" \
  --project="$PROJECT_ID" \
  --location="$REGION"

gcloud biglake iceberg catalogs create "$GCS_CATALOG" \
  --project="$PROJECT_ID" \
  --catalog-type=biglake \
  --primary-location="$REGION" \
  --default-location="$GCS_BUCKET"

将以下 Python 脚本另存为 load_gcs.py，以填充表：

import subprocess, pyarrow as pa
from pyiceberg.catalog.rest import RestCatalog
from pyiceberg.schema import Schema
from pyiceberg.types import NestedField, StringType, LongType, DoubleType
import os

PROJECT = os.environ["PROJECT_ID"]
CATALOG = os.environ["GCS_CATALOG"]
TOKEN = subprocess.check_output(
    ["gcloud", "auth", "application-default", "print-access-token"], text=True
).strip()

cat = RestCatalog(
    name=CATALOG,
    uri="https://biglake.googleapis.com/iceberg/v1/restcatalog",
    warehouse=f"bl://projects/{PROJECT}/catalogs/{CATALOG}",
    token=TOKEN,
    **{"header.x-goog-user-project": PROJECT,
       "header.X-Iceberg-Access-Delegation": "vended-credentials"},
)

cat.create_namespace_if_not_exists("retail")

def mk(name, schema, table):
    ident = ("retail", name)
    try: cat.drop_table(ident)
    except Exception: pass
    t = cat.create_table(ident, schema=schema)
    t.append(table)
    print(f"  {name}: {table.num_rows} rows")

# customers
mk("customers",
   Schema(NestedField(1, "customer_id", StringType()),
          NestedField(2, "name", StringType()),
          NestedField(3, "region", StringType())),
   pa.table({
       "customer_id": ["C1", "C2", "C3", "C4", "C5", "C6"],
       "name": ["Ana", "Ben", "Cara", "Dan", "Eve", "Finn"],
       "region": ["west", "west", "east", "east", "west", "south"],
   }))

# orders
mk("orders",
   Schema(NestedField(1, "order_id", StringType()),
          NestedField(2, "customer_id", StringType()),
          NestedField(3, "status", StringType()),
          NestedField(4, "amount", DoubleType())),
   pa.table({
       "order_id": ["O1","O2","O3","O4","O5","O6","O7","O8","O9","O10"],
       "customer_id": ["C1","C1","C2","C3","C3","C4","C5","C5","C6","C2"],
       "status": ["shipped"]*8 + ["pending","shipped"],
       "amount": [156.0,89.0,120.0,147.0,89.0,199.0,25.0,88.0,80.0,224.0],
   }))

# order_items
mk("order_items",
   Schema(NestedField(1, "order_item_id", StringType()),
          NestedField(2, "order_id", StringType()),
          NestedField(3, "product_id", StringType()),
          NestedField(4, "quantity", LongType()),
          NestedField(5, "amount", DoubleType())),
   pa.table({
       "order_item_id": [f"OI{i}" for i in range(1, 16)],
       "order_id":  ["O1","O1","O2","O3","O3","O4","O5","O6","O6","O7","O8","O9","O10","O10","O2"],
       "product_id": ["P1","P2","P3","P1","P5","P4","P3","P6","P8","P7","P1","P2","P4","P5","P6"],
       "quantity":  [1,2,1,1,3,1,1,1,4,2,1,1,1,2,1],
       "amount":    [120.0,36.0,89.0,120.0,27.0,199.0,89.0,25.0,88.0,80.0,120.0,18.0,199.0,18.0,25.0],
   }))

print("tables:", cat.list_tables("retail"))

运行脚本以加载表格：
```
python load_gcs.py
```

验证 BigQuery 中的表：

bq --location="$REGION" query --use_legacy_sql=false \
  'SELECT customer_id, name, region
   FROM `'"$PROJECT_ID"'.gcs_lake.retail.customers`
   ORDER BY customer_id'

创建 AWS Spoke

通过开放式无边界湖仓一体将 Databricks Unity Catalog 纳入 BigQuery。

在 Databricks SQL 编辑器中，创建商品表和供应商表：

CREATE SCHEMA IF NOT EXISTS `<CATALOG>`.retail;

CREATE OR REPLACE TABLE `<CATALOG>`.retail.suppliers (
  supplier_id STRING, name STRING, country STRING
) USING ICEBERG;

INSERT INTO `<CATALOG>`.retail.suppliers VALUES
  ('S1','Acme','USA'), ('S2','Globex','Germany'),
  ('S3','Initech','Japan'), ('S4','Umbrella','UK');

CREATE OR REPLACE TABLE `<CATALOG>`.retail.products (
  product_id STRING, name STRING, category STRING,
  supplier_id STRING, price DOUBLE
) USING ICEBERG;

INSERT INTO `<CATALOG>`.retail.products VALUES
  ('P1','Widget','Gadgets',  'S1',120.0),
  ('P2','Gizmo','Gadgets',   'S1', 18.0),
  ('P3','Sprocket','Parts',  'S2', 89.0),
  ('P4','Cog','Parts',       'S2',199.0),
  ('P5','Bolt','Parts',      'S3',  9.0),
  ('P6','Nut','Parts',       'S3', 25.0),
  ('P7','Gear','Machinery',  'S4', 40.0),
  ('P8','Axle','Machinery',  'S4', 22.0);

向服务主体授予读取权限：

GRANT USE CATALOG ON CATALOG `<CATALOG>` TO `<SP_APPLICATION_ID>`;
GRANT USE SCHEMA, SELECT, EXTERNAL USE SCHEMA
  ON SCHEMA `<CATALOG>`.retail
  TO `<SP_APPLICATION_ID>`;

在 Secret Manager 中存储凭据：

export CLOUDSDK_API_ENDPOINT_OVERRIDES_SECRETMANAGER="https://secretmanager.${REGION}.rep.googleapis.com/"

printf '{"client_id":"<SP_CLIENT_ID>","client_secret":"<SP_CLIENT_SECRET>"}' \
  | gcloud secrets create dbx-fed-sp \
      --project="$PROJECT_ID" \
      --location="$REGION" \
      --data-file=-

创建联合目录：

export DBX_HOST="<your-workspace-host-without-https>"
export UC_CATALOG="<CATALOG>"

gcloud alpha biglake iceberg catalogs create dbx_fed_catalog \
  --project="$PROJECT_ID