Crea un grafo de propiedades multinube en un lakehouse abierto y sin fronteras
En el siguiente instructivo, se muestra cómo compilar un solo grafo de BigQuery que unifica los silos de datos en dos nubes diferentes con un Lakehouse abierto sin fronteras y extremos de catálogo de REST de Apache Iceberg sin mover datos.
Antes de comenzar
Antes de comenzar, configura tu entorno y habilita las APIs necesarias.
Configura tu proyecto y región, y habilita las 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.comCrea un entorno virtual de Python para el cargador:
python3 -m venv iceberg-venv source iceberg-venv/bin/activate pip install --quiet "pyiceberg[pyarrow]"
Crea el radio de Google Cloud
Configurar un extremo de catálogo de REST de Apache Iceberg abierto respaldado por un bucket de Cloud Storage y cargar tres tablas de Iceberg en él
Crea el bucket y el catálogo:
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"Guarda la siguiente secuencia de comandos de Python como
load_gcs.pypara propagar las tablas: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"))Ejecuta la secuencia de comandos para cargar las tablas:
python load_gcs.pyVerifica las tablas de 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'
Crea el radio de AWS
Muestra un catálogo de Databricks Unity en BigQuery a través de un lakehouse abierto sin fronteras.
En el editor de SQL de Databricks, crea las tablas de productos y proveedores:
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);Otorga acceso de lectura a tu principal de servicio:
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>`;Almacena las credenciales en 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=-Crea el catálogo federado:
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=330sOtorga a la cuenta de servicio de BigLake acceso al 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"Verifica la sincronización:
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'
Crea el gráfico de propiedades multinube
Crea un grafo cuyas tablas de nodos y aristas abarquen ambas nubes.
Crea un conjunto de datos en tu región:
bq --location="$REGION" mk --dataset "${PROJECT_ID}:xcloud_graph"Crea el gráfico:
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 )
Consulta el gráfico multinube
Recorrer el grafo para consultar las relaciones entre ambas nubes
Ejecuta una consulta que recorra las relaciones que se encuentran en ambas nubes:
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;Ejecuta una consulta para averiguar qué países abastecen a cada cliente:
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;
Realiza una limpieza
Sigue estos pasos para evitar que se apliquen cargos a tu cuenta de Google Cloud por los recursos que usaste en esta página.
Para evitar cargos continuos, quita los recursos que creaste.
Borra el conjunto de datos de grafo:
bq rm -r -f -d "${PROJECT_ID}:xcloud_graph"Borra el catálogo federado:
gcloud alpha biglake iceberg catalogs delete dbx_fed_catalog --project="$PROJECT_ID"Guarda la siguiente secuencia de comandos como
drop_gcs.pypara descartar las tablas y el espacio de nombres de GCS: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)Ejecuta la secuencia de comandos:
python drop_gcs.pyBorra el catálogo de Cloud Storage:
gcloud biglake iceberg catalogs delete "$GCS_CATALOG" --project="$PROJECT_ID"Borra el Secret:
export CLOUDSDK_API_ENDPOINT_OVERRIDES_SECRETMANAGER="https://secretmanager.${REGION}.rep.googleapis.com/" gcloud secrets delete dbx-fed-sp --project="$PROJECT_ID" --location="$REGION"Borra el bucket:
gcloud storage rm -r "$GCS_BUCKET"
¿Qué sigue?
- Obtén más información sobre BigQuery Graph.
- Obtén más información sobre los conceptos de acceso a los datos multinube.
- Obtén información para consultar datos remotos.
- Consulta las sentencias de consulta de GQL.