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

BigQuery te permite crear y administrar tablas estándar (integradas), tablas de Apache Iceberg administradas por BigQuery y tablas externas de Apache Iceberg. Luego, puedes configurar una sesión de Apache Spark con el catálogo de tiempo de ejecución de Lakehouse para acceder a estas tablas y consultarlas directamente desde las cargas de trabajo de Managed Service para Apache Spark.

Antes de comenzar

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

  2. Habilita la API de BigQuery y la API de 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: el ID del proyecto de Google Cloud para crear el conjunto de datos.
    • DATASET_NAME: un nombre para tu conjunto de datos.

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

  3. Crea una tabla de BigQuery estándar.

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

    Reemplaza lo siguiente:

    • TABLE_NAME: Es 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 sentencia 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: un nombre para tu tabla administrada de Apache Iceberg. Por ejemplo, iceberg_managed_table
    • CONNECTION_NAME: Es el nombre de tu conexión. Creaste este recurso 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 sentencia 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 de acceso 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 de Apache Iceberg administrada por BigQuery y a tu tabla externa de Apache Iceberg.
    • CATALOG_NAME: Es el nombre del catálogo que estás usando.
    • 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 de 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 para subir 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?