Cross-Cloud-Property-Graph über ein offenes, grenzenloses Lakehouse erstellen

In der folgenden Anleitung erfahren Sie, wie Sie einen einzelnen BigQuery-Grafen erstellen, der Datensilos in zwei verschiedenen Clouds mithilfe eines offenen, grenzenlosen Lakehouse und Apache Iceberg REST-Katalogendpunkten zusammenführt, ohne Daten zu verschieben.

Hinweis

Richten Sie zuerst Ihre Umgebung ein und aktivieren Sie die erforderlichen APIs.

  1. Legen Sie Ihr Projekt und Ihre Region fest und aktivieren Sie die APIs:

    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

  2. Erstellen Sie eine virtuelle Python-Umgebung für den Loader:

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

Google Cloud -Spoke erstellen

Richten Sie einen offenen Apache Iceberg REST-Katalogendpunkt ein, der von einem Cloud Storage-Bucket unterstützt wird, und laden Sie drei Iceberg-Tabellen in diesen Bucket.

  1. Erstellen Sie den Bucket und den Katalog:

    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"

  2. Speichern Sie das folgende Python-Skript als load_gcs.py, um die Tabellen zu füllen:

    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"))

  3. Führen Sie das Script aus, um die Tabellen zu laden:

    python load_gcs.py

  4. Prüfen Sie die Tabellen aus 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 erstellen

Databricks Unity Catalog über ein offenes, grenzenloses Lakehouse in BigQuery einbinden.

  1. Erstellen Sie im Databricks SQL-Editor die Tabellen „products“ und „suppliers“:

    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);

  2. Gewähren Sie Ihrem Diensthauptkonto Lesezugriff:

    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>`;

  3. Speichern Sie die Anmeldedaten in 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=-

  4. Erstellen Sie den föderierten Katalog:

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

gcloud alpha biglake iceberg catalogs create dbx_fed_catalog \
  --project="$PROJECT_ID" \
  --catalog-type=federated \
  --federated-catalog-type=unity \
  --unity-instance-name="$DBX_HOST" \
  --unity-catalog-name="$UC_CATALOG" \
  --secret-name="projects/${PROJECT_ID}/locations/${REGION}/secrets/dbx-fed-sp" \
  --primary-location="$REGION" \
  --refresh-interval=330s

  5. Gewähren Sie dem BigLake-Dienstkonto Zugriff auf das Secret:

    export BLIRC_SA="<paste-the-blirc-...-service-account-from-the-output>"
export CLOUDSDK_API_ENDPOINT_OVERRIDES_SECRETMANAGER="https://secretmanager.${REGION}.rep.googleapis.com/"

gcloud secrets add-iam-policy-binding dbx-fed-sp \
  --project="$PROJECT_ID" \
  --location="$REGION" \
  --member="serviceAccount:${BLIRC_SA}" \
  --role="roles/secretmanager.secretAccessor"

  6. Prüfen Sie die Synchronisierung:

    bq --location="$REGION" ls dbx_fed_catalog.retail

bq --location="$REGION" query --use_legacy_sql=false \
  'SELECT product_id, name, category, supplier_id, price
   FROM `'"$PROJECT_ID"'.dbx_fed_catalog.retail.products`
   ORDER BY product_id'

Cloudübergreifenden Property-Grafen erstellen

Erstellen Sie einen Graphen, dessen Knoten- und Kantentabellen sich über beide Clouds erstrecken.

  1. Erstellen Sie ein Dataset in Ihrer Region:

    bq --location="$REGION" mk --dataset "${PROJECT_ID}:xcloud_graph"

  2. Erstellen Sie den Graphen:

    CREATE OR REPLACE PROPERTY GRAPH `PROJECT_ID.xcloud_graph.retail_xcloud`
NODE TABLES (
  `PROJECT_ID.gcs_lake.retail.customers` AS Customer
    KEY (customer_id)
    LABEL Customer PROPERTIES (customer_id, name, region),
  `PROJECT_ID.gcs_lake.retail.orders` AS Orders
    KEY (order_id)
    LABEL OrderNode PROPERTIES (order_id, status, amount),
  `PROJECT_ID.gcs_lake.retail.order_items` AS OrderItems
    KEY (order_item_id)
    LABEL OrderItem PROPERTIES (order_item_id, quantity, amount),
  `PROJECT_ID.dbx_fed_catalog.retail.products` AS Products
    KEY (product_id)
    LABEL Product PROPERTIES (product_id, name, category, price),
  `PROJECT_ID.dbx_fed_catalog.retail.suppliers` AS Suppliers
    KEY (supplier_id)
    LABEL Supplier PROPERTIES (supplier_id, name, country)
)
EDGE TABLES (
  `PROJECT_ID.gcs_lake.retail.orders` AS Placed
    KEY (order_id)
    SOURCE KEY (customer_id) REFERENCES Customer (customer_id)
    DESTINATION KEY (order_id) REFERENCES Orders (order_id)
    LABEL PLACED,
  `PROJECT_ID.gcs_lake.retail.order_items` AS Contains_Item
    KEY (order_item_id)
    SOURCE KEY (order_id) REFERENCES Orders (order_id)
    DESTINATION KEY (order_item_id) REFERENCES OrderItems (order_item_id)
    LABEL CONTAINS_ITEM,
  `PROJECT_ID.gcs_lake.retail.order_items` AS Is_Product
    KEY (order_item_id)
    SOURCE KEY (order_item_id) REFERENCES OrderItems (order_item_id)
    DESTINATION KEY (product_id) REFERENCES Products (product_id)
    LABEL IS_PRODUCT,
  `PROJECT_ID.dbx_fed_catalog.retail.products` AS Supplied_By
    KEY (product_id)
    SOURCE KEY (product_id) REFERENCES Products (product_id)
    DESTINATION KEY (supplier_id) REFERENCES Suppliers (supplier_id)
    LABEL SUPPLIED_BY
)

Cloudübergreifenden Graphen abfragen

Durchlaufen Sie den Graphen, um Beziehungen über beide Clouds hinweg abzufragen.

  1. Führen Sie eine Abfrage aus, die Beziehungen in beiden Clouds durchläuft:

    GRAPH `PROJECT_ID.xcloud_graph.retail_xcloud`
MATCH (c:Customer)-[:PLACED]->
      (o:OrderNode)-[:CONTAINS_ITEM]->
      (li:OrderItem)-[:IS_PRODUCT]->
      (p:Product)-[:SUPPLIED_BY]->
      (s:Supplier)
RETURN
  c.name AS customer,
  o.order_id AS order_id,
  p.name AS product,
  p.category AS category,
  s.name AS supplier,
  s.country AS supplier_country,
  li.quantity AS qty
ORDER BY customer, order_id
LIMIT 50;

  2. Führen Sie eine Abfrage aus, um herauszufinden, welche Länder die einzelnen Kunden beliefern:

    GRAPH `PROJECT_ID.xcloud_graph.retail_xcloud`
MATCH (c:Customer)-[:PLACED]->
      (:OrderNode)-[:CONTAINS_ITEM]->
      (:OrderItem)-[:IS_PRODUCT]->
      (:Product)-[:SUPPLIED_BY]->
      (s:Supplier)
RETURN DISTINCT c.name AS customer, s.country AS supplier_country
ORDER BY customer, supplier_country;

Bereinigen

Mit den folgenden Schritten vermeiden Sie, dass Ihrem Google Cloud -Konto die auf dieser Seite verwendeten Ressourcen in Rechnung gestellt werden:

Entfernen Sie die von Ihnen erstellten Ressourcen, um laufende Kosten zu vermeiden.

  1. Löschen Sie das Graph-Dataset:

    bq rm -r -f -d "${PROJECT_ID}:xcloud_graph"

  2. Löschen Sie den föderierten Katalog:

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

  3. Speichern Sie das folgende Script als drop_gcs.py, um die GCS-Tabellen und den Namespace zu löschen:

    import subprocess, os
from pyiceberg.catalog.rest import RestCatalog

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},
)
for tbl in cat.list_tables("retail"):
    cat.drop_table(tbl)
cat.drop_namespace("retail")
print("emptied", CATALOG)

  4. Führen Sie das Script aus:

    python drop_gcs.py

  5. Löschen Sie den Cloud Storage-Katalog:

    gcloud biglake iceberg catalogs delete "$GCS_CATALOG" --project="$PROJECT_ID"

  6. Löschen Sie das Secret:

    export CLOUDSDK_API_ENDPOINT_OVERRIDES_SECRETMANAGER="https://secretmanager.${REGION}.rep.googleapis.com/"
gcloud secrets delete dbx-fed-sp --project="$PROJECT_ID" --location="$REGION"

  7. Löschen Sie den Bucket:

    gcloud storage rm -r "$GCS_BUCKET"

Nächste Schritte