Funcionalidades adicionais da metastore do BigLake

Para personalizar a configuração do metastore do BigLake, pode usar as seguintes funcionalidades adicionais:

  • Procedimentos do Apache Spark Iceberg
  • A opção de filtro para tabelas não suportadas
  • Substituições da ligação do BigQuery
  • Políticas de controlo de acesso para tabelas Iceberg do metastore do BigLake

Use procedimentos do Iceberg Spark

Para usar procedimentos do Iceberg Spark, tem de incluir extensões SQL do Iceberg na sua configuração do Spark. Por exemplo, pode criar um procedimento para reverter para um estado anterior.

Use o Spark-SQL interativo para reverter para um estado anterior

Pode usar um procedimento do Iceberg Spark para criar, modificar e reverter uma tabela para o respetivo estado anterior. Por exemplo:

  1. Criar uma tabela do 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,gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.6.1-1.0.1-beta.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

    Substitua o seguinte:

    • CATALOG_NAME: o nome do catálogo que faz referência à sua tabela do Spark.

    • PROJECT_ID: o ID do Google Cloud projeto.

    • WAREHOUSE_DIRECTORY: o URI da pasta do Cloud Storage onde o seu data warehouse está armazenado.

    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;

    Substitua o seguinte:

    • NAMESPACE_NAME: o nome do espaço de nomes que faz referência à sua tabela do Spark.
    • TABLE_NAME: um nome de tabela que faz referência à sua tabela do Spark.

    A saída contém detalhes sobre a configuração da tabela:

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

  2. Altere novamente a tabela e, em seguida, reverta-a para a cópia instantânea criada anteriormente 1659239298328512231:

    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;

    Substitua o seguinte:

    • SNAPSHOT_ID: o ID do resumo para o qual está a reverter.

    O resultado é semelhante ao seguinte:

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

Filtre tabelas não suportadas de funções de listagem de tabelas

Quando usa o Spark SQL com o catálogo do metastore do BigLake, o comando SHOW TABLES mostra todas as tabelas no espaço de nomes especificado, mesmo as que não são compatíveis com o Spark.

Para apresentar apenas as tabelas suportadas, ative a opção 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,gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.6.1-1.0.1-beta.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"

Substitua o seguinte:

  • CATALOG_NAME: o nome do catálogo do Spark a usar.
  • PROJECT_ID: o ID do Google Cloud projeto a usar.
  • LOCATION: a localização dos recursos do BigQuery.
  • WAREHOUSE_DIRECTORY: a pasta do Cloud Storage a usar como armazém de dados.

Defina uma substituição da ligação do BigQuery

Pode usar ligações do BigQuery para aceder a dados armazenados fora do BigQuery, como no Cloud Storage.

Para definir uma substituição da associação do BigQuery que faculte acesso a um contentor do Cloud Storage, conclua os seguintes passos:

  1. No seu projeto do BigQuery, crie uma nova ligação ao seu recurso do Cloud Storage. Esta associação define a forma como o BigQuery acede aos seus dados.

  2. Conceda à conta de utilizador ou de serviço que acede aos dados a função roles/bigquery.connectionUser na ligação.

    Certifique-se de que o recurso de ligação partilha a mesma localização que os recursos de destino no BigQuery. Para mais informações, consulte o artigo Faça a gestão das associações.

  3. Especifique a associação na sua tabela Iceberg com a propriedade 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');

    Substitua o seguinte:

    • TABLE_NAME: um nome de tabela para a sua tabela do Spark.
    • WAREHOUSE_DIRECTORY: o URI do contentor do Cloud Storage que armazena os seus dados.
    • PROJECT_ID: o ID do Google Cloud projeto a usar.
    • LOCATION: a localização da ligação.
    • CONNECTION_ID: o ID da associação.

Defina políticas de controlo de acesso

Pode ativar o controlo de acesso detalhado (FGAC) em tabelas Iceberg do metastore do BigLake configurando políticas de controlo de acesso. Só pode definir políticas de controlo de acesso em tabelas que usam uma substituição da ligação ao BigQuery. Pode definir estas políticas das seguintes formas:

Depois de configurar as políticas de FGAC, pode consultar a tabela a partir do Spark através do seguinte exemplo:

from pyspark.sql import SparkSession

# Create a Spark session
spark = SparkSession.builder \
.appName("BigLake Metastore 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()

Substitua o seguinte:

  • CATALOG_NAME: o nome do seu catálogo.
  • PROJECT_ID: o ID do projeto que contém os seus recursos do BigQuery.
  • LOCATION: a localização dos recursos do BigQuery.
  • WAREHOUSE_DIRECTORY: o URI da pasta do Cloud Storage que contém o seu armazém de dados.
  • MATERIALIZATION_NAMESPACE: o espaço de nomes onde quer armazenar os resultados temporários.
  • DATASET_NAME: o nome do conjunto de dados que contém a tabela que está a consultar.
  • ICEBERG_TABLE_NAME: o nome da tabela que está a consultar.

O que se segue?