Usa el catálogo del entorno de ejecución de Lakehouse con tablas en BigQuery

En este documento, se proporciona un ejemplo de cómo usar el catálogo de tiempo de ejecución de Lakehouse con tablas de BigQuery y Managed Service para Apache Spark.

Con el catálogo del entorno de ejecución de Lakehouse, puedes crear y usar tablas estándar (integradas), tablas de Apache Iceberg administradas por BigQuery y tablas externas de Apache Iceberg desde BigQuery.

Antes de comenzar

  1. Habilita la facturación para tu proyecto de Google Cloud . Obtén información para verificar si la facturación está habilitada en un proyecto.

  2. Habilita las APIs de BigQuery y Dataproc.

    Habilitar las API

Roles obligatorios

Para obtener los permisos que necesitas para usar Managed Service para Apache Spark con el catálogo de tiempo de ejecución de Lakehouse como almacén de metadatos, pídele a tu administrador que te otorgue los siguientes roles de IAM:

Para obtener más información sobre cómo otorgar roles, consulta Administra el acceso a proyectos, carpetas y organizaciones.

También puedes obtener los permisos necesarios a través de roles personalizados o cualquier otro rol predefinido.

Conectarse a una tabla

  1. Crea un conjunto de datos en la consola de Google Cloud .

    CREATE SCHEMA `PROJECT_ID`.DATASET_NAME;

    Reemplaza lo siguiente:

    • PROJECT_ID: Es el ID del proyecto Google Cloud en el que se creará el conjunto de datos.
    • DATASET_NAME: Es un nombre para tu conjunto de datos.

  2. Crea una conexión de recursos de Cloud.

  3. Crea una tabla estándar de BigQuery.

    CREATE TABLE `PROJECT_ID`.DATASET_NAME.TABLE_NAME (name STRING,id INT64);

    Reemplaza lo siguiente:

    • TABLE_NAME: Un nombre para tu tabla.

  4. Inserta datos en la tabla estándar de BigQuery.

    INSERT INTO `PROJECT_ID`.DATASET_NAME.TABLE_NAME VALUES ('test_name1', 123),('test_name2', 456),('test_name3', 789);

  5. Crea una tabla de Apache Iceberg administrada por BigQuery.

    Por ejemplo, para crear una tabla, ejecuta la siguiente instrucción CREATE.

    CREATE TABLE `PROJECT_ID`.DATASET_NAME.ICEBERG_TABLE_NAME(
name STRING,id INT64
)
WITH CONNECTION `CONNECTION_NAME`
OPTIONS (
file_format = 'PARQUET',
table_format = 'ICEBERG',
storage_uri = 'STORAGE_URI');

    Reemplaza lo siguiente:

    • ICEBERG_TABLE_NAME: Es el nombre de tu tabla administrada de Apache Iceberg. Por ejemplo, iceberg_managed_table
    • CONNECTION_NAME: Es el nombre de tu conexión. La creaste en el paso anterior. Por ejemplo, myproject.us.myconnection
    • STORAGE_URI: Es un URI de Cloud Storage completamente calificado. Por ejemplo, gs://mybucket/table

  6. Inserta datos en la tabla de Apache Iceberg administrada por BigQuery.

    INSERT INTO `PROJECT_ID`.DATASET_NAME.ICEBERG_TABLE_NAME VALUES ('test_name1', 123),('test_name2', 456),('test_name3', 789);

  7. Crea una tabla externa de Apache Iceberg.

    Por ejemplo, para crear una tabla externa de Apache Iceberg, ejecuta la siguiente declaración CREATE.

    CREATE OR REPLACE EXTERNAL TABLE  `PROJECT_ID`.DATASET_NAME.READONLY_ICEBERG_TABLE_NAME
WITH CONNECTION `CONNECTION_NAME`
OPTIONS (
  format = 'ICEBERG',
  uris =
    ['BUCKET_PATH'],
  require_partition_filter = FALSE);

    Reemplaza lo siguiente:

    • READONLY_ICEBERG_TABLE_NAME: Es un nombre para tu tabla de solo lectura.
    • BUCKET_PATH: Es la ruta al bucket de Cloud Storage que contiene los datos de la tabla externa, en el formato ['gs://bucket_name/[folder_name/]file_name'].

  8. Desde Apache Spark, consulta la tabla estándar, la tabla de Apache Iceberg administrada por BigQuery y la tabla externa de Apache Iceberg.

    from pyspark.sql import SparkSession

# Create a spark session
spark = SparkSession.builder \
.appName("Lakehouse runtime catalog Iceberg") \
.config("spark.sql.catalog.CATALOG_NAME", "org.apache.iceberg.spark.SparkCatalog") \
.config("spark.sql.catalog.CATALOG_NAME.catalog-impl", "org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog") \
.config("spark.sql.catalog.CATALOG_NAME.gcp_project", "PROJECT_ID") \
.config("spark.sql.catalog.CATALOG_NAME.gcp_location", "LOCATION") \
.config("spark.sql.catalog.CATALOG_NAME.warehouse", "WAREHOUSE_DIRECTORY") \
.getOrCreate()
spark.conf.set("viewsEnabled","true")

# Use the Lakehouse runtime catalog
spark.sql("USE `CATALOG_NAME`;")
spark.sql("USE NAMESPACE DATASET_NAME;")

# Configure spark for temp results
spark.sql("CREATE namespace if not exists MATERIALIZATION_NAMESPACE");
spark.conf.set("materializationDataset","MATERIALIZATION_NAMESPACE")

# List the tables in the dataset
df = spark.sql("SHOW TABLES;")
df.show();

# Query the tables
sql = """SELECT * FROM DATASET_NAME.TABLE_NAME"""
df = spark.read.format("bigquery").load(sql)
df.show()

sql = """SELECT * FROM DATASET_NAME.ICEBERG_TABLE_NAME"""
df = spark.read.format("bigquery").load(sql)
df.show()

sql = """SELECT * FROM DATASET_NAME.READONLY_ICEBERG_TABLE_NAME"""
df = spark.read.format("bigquery").load(sql)
df.show()

    Reemplaza lo siguiente:

    • WAREHOUSE_DIRECTORY: Es el URI de la carpeta de Cloud Storage que está conectada a tu tabla externa de Apache Iceberg y a tu tabla administrada de Apache Iceberg de BigQuery.
    • CATALOG_NAME: Es el nombre del catálogo que usas.
    • MATERIALIZATION_NAMESPACE: Es el espacio de nombres para almacenar los resultados temporales.

  9. Ejecuta la secuencia de comandos de Apache Spark con Managed Service para Apache Spark.

    gcloud dataproc batches submit pyspark SCRIPT_PATH \
  --version=2.2 \
  --project=PROJECT_ID \
  --region=REGION \
  --deps-bucket=YOUR_BUCKET \

    Reemplaza lo siguiente:

    • SCRIPT_PATH: Es la ruta de acceso a la secuencia de comandos que usa el trabajo por lotes.
    • PROJECT_ID: Es el ID del proyecto Google Cloud en el que se ejecutará el trabajo por lotes.
    • REGION: Es la región en la que se ejecuta tu carga de trabajo.
    • YOUR_BUCKET: Es la ubicación del bucket de Cloud Storage en el que se subirán las dependencias de la carga de trabajo. No se requiere el prefijo de URI gs:// del bucket. Puedes especificar la ruta de acceso o el nombre del bucket, por ejemplo, mybucketname1.

¿Qué sigue?