Funciones adicionales del catálogo de entorno de ejecución de Lakehouse

Para personalizar la configuración de tu catálogo de entorno de ejecución de Lakehouse, puedes usar las siguientes funciones adicionales:

  • Son procedimientos de Apache Spark Iceberg.
  • Es la opción de filtro para las tablas no admitidas.
  • Son los reemplazos de la conexión de BigQuery.
  • Son políticas de control de acceso para las tablas de Iceberg del catálogo de entornos de ejecución de Lakehouse.

Usa procedimientos de Apache Iceberg Spark

Para usar procedimientos de Apache Spark, debes incluir extensiones de SQL de Apache Iceberg en tu configuración de Apache Spark. Por ejemplo, puedes crear un procedimiento para revertir a un estado anterior.

Usa Apache Spark SQL interactivo para revertir a un estado anterior

Puedes usar un procedimiento de Apache Spark para crear, modificar y revertir una tabla a su estado anterior. Por ejemplo:

  1. Crea una tabla de Apache Spark:

    spark-sql \
   --jars https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-spark-runtime-3.5_2.12/1.6.1/iceberg-spark-runtime-3.5_2.12-1.6.1.jar,BIGLAKE_ICEBERG_CATALOG_JAR \
   --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
   --conf spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog \
   --conf spark.sql.catalog.CATALOG_NAME.catalog-impl=org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog \
   --conf spark.sql.catalog.CATALOG_NAME.gcp_project=PROJECT_ID \
   --conf spark.sql.catalog.CATALOG_NAME.warehouse=WAREHOUSE_DIRECTORY

    Reemplaza lo siguiente:

    • BIGLAKE_ICEBERG_CATALOG_JAR: Es el URI de Cloud Storage del complemento del catálogo personalizado de Apache Iceberg que se usará. Según el número de versión de Apache Iceberg, selecciona una de las siguientes opciones:
      • Iceberg 1.9.1: gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.9.1-1.0.1.jar
      • Iceberg 1.6.1: gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.6.1-1.0.2.jar
    • CATALOG_NAME: Es el nombre del catálogo que hace referencia a tu tabla de Apache Spark.

    • PROJECT_ID: Es el ID del proyecto de Google Cloud .

    • WAREHOUSE_DIRECTORY: Es el URI de la carpeta de Cloud Storage en la que se almacena tu data warehouse.

    USE `CATALOG_NAME`;
CREATE NAMESPACE NAMESPACE_NAME;
USE NAMESPACE NAMESPACE_NAME;
CREATE TABLE NAMESPACE_NAME.TABLE_NAME (id int, data string) USING ICEBERG LOCATION 'WAREHOUSE_DIRECTORY';
INSERT INTO NAMESPACE_NAME.TABLE_NAME VALUES (1, "first row");
DESCRIBE EXTENDED TABLE_NAME;

    Reemplaza lo siguiente:

    • NAMESPACE_NAME: Es el nombre del espacio de nombres que hace referencia a tu tabla de Apache Spark.
    • TABLE_NAME: Es el nombre de una tabla que hace referencia a tu tabla de Apache Spark.

    El resultado contiene detalles sobre la configuración de la tabla:

    ...
Table Properties [current-snapshot-id=1659239298328512231,format=iceberg/parquet,format-version=2,write.parquet.compression-codec=zstd]
...

  2. Vuelve a modificar la tabla y, luego, revierte a la instantánea 1659239298328512231 creada anteriormente:

    ALTER TABLE TABLE_NAME ADD COLUMNS (newDoubleCol double);
INSERT INTO TABLE_NAME VALUES (2, "second row", 2.5);
SELECT * FROM TABLE_NAME;
CALL CATALOG_NAME.system.set_current_snapshot('NAMESPACE_NAME.TABLE_NAME', SNAPSHOT_ID);
SELECT * FROM TABLE_NAME;

    Reemplaza lo siguiente:

    • SNAPSHOT_ID: Es el ID de la instantánea a la que reviertes.

    El resultado es similar a lo siguiente:

    1 first row
Time taken: 0.997 seconds, Fetched 1 row(s)

Filtrar tablas no compatibles de las funciones de enumeración de tablas

Cuando usas Apache Spark SQL con el catálogo de tiempo de ejecución de Lakehouse, el comando SHOW TABLES muestra todas las tablas en el espacio de nombres especificado, incluso las que no son compatibles con Apache Spark.

Para mostrar solo las tablas compatibles, activa la opción filter_unsupported_tables:

spark-sql
  --jars https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-spark-runtime-3.5_2.12/1.6.1/iceberg-spark-runtime-3.5_2.12-1.6.1.jar,BIGLAKE_ICEBERG_CATALOG_JAR \
  --conf spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog \
  --conf spark.sql.catalog.CATALOG_NAME.catalog-impl=org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog \
  --conf spark.sql.catalog.CATALOG_NAME.gcp_project=PROJECT_ID \
  --conf spark.sql.catalog.CATALOG_NAME.gcp_location=LOCATION \
  --conf spark.sql.catalog.CATALOG_NAME.warehouse=WAREHOUSE_DIRECTORY \
  --conf spark.sql.catalog.CATALOG_NAME.filter_unsupported_tables="true"

Reemplaza lo siguiente:

  • BIGLAKE_ICEBERG_CATALOG_JAR: Es el URI de Cloud Storage del complemento del catálogo personalizado de Apache Iceberg que se usará. Según el número de versión de Apache Iceberg, selecciona una de las siguientes opciones:
    • Iceberg 1.9.1: gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.9.1-1.0.1.jar
    • Iceberg 1.6.1: gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.6.1-1.0.2.jar
  • CATALOG_NAME: Es el nombre del catálogo de Apache Spark que se usará.
  • PROJECT_ID: Es el ID del proyecto de Google Cloud que se usará.
  • LOCATION: Es la ubicación de los recursos de BigQuery.
  • WAREHOUSE_DIRECTORY: Es la carpeta de Cloud Storage que se usará como almacén de datos.

Cómo establecer una anulación de conexión de BigQuery

Puedes usar conexiones de BigQuery para acceder a datos almacenados fuera de BigQuery, como en Cloud Storage.

Para establecer una anulación de conexión de BigQuery que proporcione acceso a un bucket de Cloud Storage, completa los siguientes pasos:

  1. En tu proyecto de BigQuery, crea una conexión nueva a tu recurso de Cloud Storage. Esta conexión define cómo BigQuery accede a tus datos.

  2. Otorga a la cuenta de usuario o de servicio que accede a los datos el rol roles/bigquery.connectionUser en la conexión.

    Asegúrate de que el recurso de conexión comparta la misma ubicación que los recursos de destino en BigQuery. Para obtener más información, consulta Administra conexiones.

  3. Especifica la conexión en tu tabla de Apache Iceberg con la propiedad bq_connection:

    CREATE TABLE TABLE_NAME (id int, data string) USING ICEBERG LOCATION 'WAREHOUSE_DIRECTORY' TBLPROPERTIES ('bq_connection'='projects/PROJECT_ID/locations/LOCATION/connections/CONNECTION_ID');

    Reemplaza lo siguiente:

    • TABLE_NAME: Es el nombre de la tabla de Apache Spark.
    • WAREHOUSE_DIRECTORY: Es el URI del bucket de Cloud Storage que almacena tus datos.
    • PROJECT_ID: Es el ID del proyecto de Google Cloud que se usará.
    • LOCATION: Es la ubicación de la conexión.
    • CONNECTION_ID: el ID de la conexión.

Configura las políticas de control de acceso

Puedes habilitar el control de acceso detallado (FGAC) en las tablas de Apache Iceberg del catálogo del entorno de ejecución de Lakehouse configurando políticas de control de acceso. Solo puedes establecer políticas de control de acceso en tablas que usan una anulación de conexión de BigQuery. Puedes establecer estas políticas de las siguientes maneras:

Después de configurar tus políticas de FGAC, puedes consultar la tabla desde Apache Spark con el siguiente ejemplo:

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.sql("USE `CATALOG_NAME`;")

# Configure spark for storing temp results
spark.conf.set("viewsEnabled","true")
spark.sql("CREATE NAMESPACE IF NOT EXISTS MATERIALIZATION_NAMESPACE");
spark.conf.set("materializationDataset","MATERIALIZATION_NAMESPACE")

spark.sql("USE NAMESPACE DATASET_NAME;")

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

Reemplaza lo siguiente:

  • CATALOG_NAME: Es el nombre de tu catálogo.
  • PROJECT_ID: Es el ID del proyecto que contiene tus recursos de BigQuery.
  • LOCATION: La ubicación de los recursos de BigQuery.
  • WAREHOUSE_DIRECTORY: Es el URI de la carpeta de Cloud Storage que contiene tu almacén de datos.
  • MATERIALIZATION_NAMESPACE: Es el espacio de nombres en el que deseas almacenar los resultados temporales.
  • DATASET_NAME: Es el nombre del conjunto de datos que contiene la tabla que consultas.
  • ICEBERG_TABLE_NAME: Es el nombre de la tabla que consultas.

¿Qué sigue?