Use o metastore do BigLake com o catálogo REST do Iceberg

O catálogo REST do Apache Iceberg gerido no metastore do BigLake cria interoperabilidade entre todos os seus motores de consulta, oferecendo uma única origem de verdade para todos os seus dados do Iceberg. Permite que os motores de consulta, como o Apache Spark, descubram, leiam metadados e façam a gestão de tabelas Iceberg de forma consistente.

As tabelas Iceberg que usa com o catálogo REST do Iceberg são denominadas tabelas BigLake para o Apache Iceberg (pré-visualização). Estas são tabelas Iceberg que cria a partir de motores de código aberto e armazena no Cloud Storage. Podem ser lidos por motores de código aberto ou pelo BigQuery. As gravações só são suportadas a partir de motores de código aberto. Neste documento, referimo-nos a estas tabelas como tabelas Iceberg do BigLake.

Antes de começar

  1. Verify that billing is enabled for your Google Cloud project.

    Saiba como verificar se a faturação está ativada num projeto.

  2. Enable the BigLake API.

    Roles required to enable APIs

    To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

    Enable the API

  3. Opcional: compreenda como funciona o metastore do BigLake e por que motivo o deve usar.

Funções necessárias

Para receber as autorizações de que precisa para usar o catálogo REST do Iceberg no metastore do BigLake, peça ao seu administrador que lhe conceda as seguintes funções de IAM no seu projeto ou conta de serviço:

Para mais informações sobre a atribuição de funções, consulte o artigo Faça a gestão do acesso a projetos, pastas e organizações.

Também pode conseguir as autorizações necessárias através de funções personalizadas ou outras funções predefinidas.

Limitações

O catálogo REST do Iceberg está sujeito às seguintes limitações:

Configure um motor de consultas com o catálogo REST do Iceberg

Iceberg 1.10 ou posterior

O Iceberg 1.10 e as versões subsequentes têm suporte integrado para fluxos de autorização da Google no GoogleAuthManager. O Dataproc Spark também suporta GoogleAuthManager nas seguintes versões:

  • Versões de tempo de execução 2.2 do Dataproc no Compute Engine 2.2.65 e posteriores
  • Sem servidor para imagens do Apache Spark 2.2 2.2.60 e posteriores
  • Versões de tempo de execução 2.3 do Dataproc no Compute Engine 2.3.11 e posteriores
  • Imagens 2.3.10 e posteriores do Apache Spark 2.3 sem servidor
import pyspark
from pyspark.context import SparkContext
from pyspark.sql import SparkSession

catalog_name = "CATALOG_NAME"
spark = SparkSession.builder.appName("APP_NAME") \
  .config(f'spark.sql.catalog.{catalog_name}', 'org.apache.iceberg.spark.SparkCatalog') \
  .config(f'spark.sql.catalog.{catalog_name}.type', 'rest') \
  .config(f'spark.sql.catalog.{catalog_name}.uri', 'https://biglake.googleapis.com/iceberg/v1beta/restcatalog') \
  .config(f'spark.sql.catalog.{catalog_name}.warehouse', 'gs://STORAGE_URI') \
  .config(f'spark.sql.catalog.{catalog_name}.header.x-goog-user-project', 'PROJECT_ID') \
  .config(f'spark.sql.catalog.{catalog_name}.rest.auth.type', 'org.apache.iceberg.gcp.auth.GoogleAuthManager') \
  .config(f'spark.sql.catalog.{catalog_name}.io-impl', 'org.apache.iceberg.hadoop.HadoopFileIO') \
  .config(f'spark.sql.catalog.{catalog_name}.rest-metrics-reporting-enabled', 'false') \
  .config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions') \
  .config('spark.sql.defaultCatalog', 'CATALOG_NAME') \
  .getOrCreate()

Substitua o seguinte:

  • CATALOG_NAME: um nome para o seu catálogo REST do Iceberg.
  • APP_NAME: um nome para a sua sessão do Spark.
  • STORAGE_URI: o URI do contentor do Cloud Storage que contém as suas tabelas Iceberg do BigLake.
  • PROJECT_ID: o projeto que é faturado pela utilização do catálogo REST do Iceberg, que pode ser diferente do projeto proprietário do contentor do Cloud Storage. Para ver detalhes sobre a configuração do projeto quando usa uma API REST, consulte os Parâmetros do sistema.

Lançamentos anteriores do Iceberg

Para lançamentos do Iceberg anteriores à versão 1.10 que não estão incluídos nas imagens do Dataproc, pode configurar a autenticação OAuth padrão configurando uma sessão com o seguinte:

import pyspark
from pyspark.context import SparkContext
from pyspark.sql import SparkSession

catalog_name = "CATALOG_NAME"
spark = SparkSession.builder.appName("APP_NAME") \
  .config(f'spark.sql.catalog.{catalog_name}', 'org.apache.iceberg.spark.SparkCatalog') \
  .config(f'spark.sql.catalog.{catalog_name}.type', 'rest') \
  .config(f'spark.sql.catalog.{catalog_name}.uri', 'https://biglake.googleapis.com/iceberg/v1beta/restcatalog') \
  .config(f'spark.sql.catalog.{catalog_name}.warehouse', 'gs://STORAGE_URI') \
  .config(f'spark.sql.catalog.{catalog_name}.header.x-goog-user-project', 'PROJECT_ID') \
  .config(f"spark.sql.catalog.{catalog_name}.token", "TOKEN") \
  .config(f"spark.sql.catalog.{catalog_name}.oauth2-server-uri", "https://oauth2.googleapis.com/token") \
  .config(f'spark.sql.catalog.{catalog_name}.io-impl', 'org.apache.iceberg.hadoop.HadoopFileIO') \
  .config(f'spark.sql.catalog.{catalog_name}.rest-metrics-reporting-enabled', 'false') \
  .config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions') \
  .config('spark.sql.defaultCatalog', 'CATALOG_NAME') \
  .getOrCreate()

Substitua o seguinte:

  • CATALOG_NAME: um nome para o seu catálogo REST do Iceberg.
  • APP_NAME: um nome para a sua sessão do Spark.
  • STORAGE_URI: o URI do contentor do Cloud Storage que contém as suas tabelas Iceberg do BigLake.
  • PROJECT_ID: o projeto que é faturado pela utilização do catálogo REST do Iceberg, que pode ser diferente do projeto proprietário do contentor do Cloud Storage. Para ver detalhes sobre a configuração do projeto quando usa uma API REST, consulte os Parâmetros do sistema.
  • TOKEN: o seu token de autenticação, que é válido durante uma hora. Por exemplo, um token gerado através de gcloud auth application-default print-access-token.

Crie um espaço de nomes

spark.sql("CREATE NAMESPACE IF NOT EXISTS NAMESPACE_NAME;")

spark.sql("USE NAMESPACE_NAME;")

Substitua NAMESPACE_NAME por um nome para o seu espaço de nomes.

Criar uma tabela

spark.sql("CREATE TABLE TABLE_NAME (id int, data string) USING ICEBERG;")

spark.sql("DESCRIBE NAMESPACE_NAME.TABLE_NAME").show()

Substitua o seguinte:

  • NAMESPACE_NAME: o nome do seu espaço de nomes
  • TABLE_NAME: um nome para a tabela

Listar tabelas

spark.sql("SHOW TABLES").show()

Insira dados na tabela

O exemplo seguinte insere dados de exemplo na tabela:

spark.sql("INSERT INTO TABLE_NAME VALUES (1, \"first row\"), (2, \"second row\"), (3, \"third row\");")

Consultar uma tabela

O exemplo seguinte seleciona todos os dados da tabela:

spark.sql("SELECT * FROM TABLE_NAME;").show()

O exemplo seguinte consulta a mesma tabela do BigQuery:

SELECT * FROM `CLOUD_STORAGE_BUCKET_NAME>NAMESPACE_NAME.TABLE_NAME`;

Substitua CLOUD_STORAGE_BUCKET_NAME pelo nome do contentor do Cloud Storage para o seu catálogo REST do Iceberg. Por exemplo, se o URI for gs://iceberg_bucket, use iceberg_bucket.

Altere um esquema de tabela

O exemplo seguinte adiciona uma coluna à tabela:

spark.sql("ALTER TABLE TABLE_NAME ADD COLUMNS ( desc string);")
spark.sql("DESCRIBE NAMESPACE_NAME.TABLE_NAME").show()

Elimine uma tabela

O exemplo seguinte elimina a tabela do espaço de nomes indicado:

spark.sql("DROP TABLE TABLE_NAME;")

Preços

Para ver detalhes de preços, consulte os preços do BigLake.

O que se segue?