Usar a metastore do BigQuery com tabelas no BigQuery

Este documento explica como usar a metastore do BigQuery com tabelas do BigQuery e o Spark.

Com o metastore do BigQuery, é possível criar e usar tabelas padrão (integradas), tabelas do BigQuery para Apache Spark Iceberg e tabelas externas do BigLake no BigQuery.

Antes de começar

  1. Ative o faturamento do seu Google Cloud projeto. Saiba como verificar se o faturamento está ativado em um projeto.

  2. Ative as APIs BigQuery e Dataproc.

    Ativar as APIs

  3. Opcional: entenda como a metastore do BigQuery funciona e por que você deve usá-la.

Funções exigidas

Para receber as permissões necessárias para usar o Spark e o Dataproc com o metastore do BigQuery como uma loja de metadados, peça ao administrador para conceder a você os seguintes papéis do IAM:

Para mais informações sobre a concessão de papéis, consulte Gerenciar o acesso a projetos, pastas e organizações.

Também é possível conseguir as permissões necessárias por meio de papéis personalizados ou de outros papéis predefinidos.

Conectar-se a uma tabela

  1. Crie um conjunto de dados no console Google Cloud .

    CREATE SCHEMA `PROJECT_ID`.DATASET_NAME;

    Substitua:

    • PROJECT_ID: o ID do projeto do Google Cloud para criar o conjunto de dados.
    • DATASET_NAME: um nome para o conjunto de dados.

  2. Crie uma conexão de recurso do Cloud.

  3. Crie uma tabela padrão do BigQuery.

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

    Substitua:

    • TABLE_NAME: um nome para a tabela.

  4. Insira dados na tabela padrão do BigQuery.

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

  5. Crie uma tabela do BigQuery para o Apache Iceberg.

    Por exemplo, para criar uma tabela, execute a instrução CREATE a seguir.

    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');

    Substitua:

    • ICEBERG_TABLE_NAME: um nome para a tabela Iceberg. Por exemplo, iceberg_managed_table.
    • CONNECTION_NAME: o nome da conexão. Você criou isso na etapa anterior. Por exemplo, myproject.us.myconnection.
    • STORAGE_URI: um URI do Cloud Storage totalmente qualificado. Por exemplo, gs://mybucket/table.

  6. Insira dados na tabela do BigQuery para o Apache Iceberg.

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

  7. Crie uma tabela Iceberg somente leitura.

    Por exemplo, para criar uma tabela Iceberg somente leitura, execute a instrução CREATE abaixo.

    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);

    Substitua:

    • READONLY_ICEBERG_TABLE_NAME: um nome para a tabela somente leitura.
    • BUCKET_PATH: o caminho para o bucket do Cloud Storage que contém os dados da tabela externa, no formato ['gs://bucket_name/[folder_name/]file_name'].

  8. No PySpark, consulte a tabela padrão, a tabela Iceberg gerenciada e a tabela Iceberg somente leitura.

    from pyspark.sql import SparkSession

# Create a spark session
spark = SparkSession.builder \
.appName("BigQuery 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.conf.set("viewsEnabled","true")

# Use the bqms_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 a standard BigQuery table
sql = """SELECT * FROM DATASET_NAME.TABLE_NAME"""
df = spark.read.format("bigquery").load(sql)
df.show()

# Query a BigQuery Managed Apache Iceberg table
sql = """SELECT * FROM DATASET_NAME.ICEBERG_TABLE_NAME"""
df = spark.read.format("bigquery").load(sql)
df.show()

# Query a BigQuery Readonly Apache Iceberg table
sql = """SELECT * FROM DATASET_NAME.READONLY_ICEBERG_TABLE_NAME"""
df = spark.read.format("bigquery").load(sql)
df.show()

    Substitua:

    • WAREHOUSE_DIRECTORY: o URI da pasta do Cloud Storage que contém o data warehouse.
    • CATALOG_NAME: o nome do catálogo que você está usando.
    • MATERIALIZATION_NAMESPACE: o namespace para armazenar resultados temporários.

  9. Execute o script PySpark usando o Spark sem servidor.

    gcloud dataproc batches submit pyspark SCRIPT_PATH \
  --version=2.2 \
  --project=PROJECT_ID \
  --region=REGION \
  --deps-bucket=YOUR_BUCKET \
  --jars=https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-spark-runtime-3.5_2.12/1.5.2/iceberg-spark-runtime-3.5_2.12-1.5.2.jar,gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.5.2-1.0.0-beta.jar

    Substitua:

    • SCRIPT_PATH: o caminho para o script usado pelo job em lote.
    • PROJECT_ID: o ID do projeto Google Cloud para executar o job em lote.
    • REGION: a região em que a carga de trabalho é executada.
    • YOUR_BUCKET: o local do bucket do Cloud Storage para fazer upload das dependências da carga de trabalho. O prefixo URI gs:// do bucket não é necessário. É possível especificar o caminho ou o nome do bucket, por exemplo, mybucketname1.

A seguir