Usar a metastore do BigQuery com o Dataproc

Neste documento, explicamos como usar a metastore do BigQuery com o Dataproc no Compute Engine. Essa conexão oferece um metastore único e compartilhado que funciona em mecanismos de software de código aberto, como o Apache Spark.

Antes de começar

  1. Ative o faturamento para o projeto do Google Cloud. 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 a 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.

Fluxo de trabalho geral

Para usar o Dataproc no Compute Engine com a metastore do BigQuery, siga estas etapas gerais:

  1. Crie ou configure um cluster do Dataproc.
  2. Conecte-se ao mecanismo de software de código aberto que preferir, como o Spark.
  3. Use um arquivo JAR para instalar o plug-in de catálogo do Apache Iceberg no cluster.
  4. Crie e gerencie seus recursos do Metastore do BigQuery conforme necessário, dependendo do mecanismo de software de código aberto que você está usando.
  5. No BigQuery, acesse e use seus recursos do Metastore do BigQuery.

Conectar a metastore do BigQuery ao Spark

As instruções a seguir mostram como conectar o Dataproc à metastore do BigQuery usando o Spark SQL interativo.

Fazer o download do plug-in do catálogo Iceberg

Para conectar a metastore do BigQuery ao Dataproc e ao Spark, use o arquivo jar do plug-in do catálogo Iceberg da metastore do BigQuery.

Esse arquivo é incluído por padrão na versão 2.2 da imagem do Dataproc. Se os clusters do Dataproc não tiverem acesso direto à Internet, faça o download do plug-in e faça upload dele para um bucket do Cloud Storage que o cluster do Dataproc possa acessar.

Faça o download do plug-in do catálogo Apache Iceberg para a metastore do BigQuery.

Configurar um cluster do Dataproc

Antes de se conectar ao metastore do BigQuery, é necessário configurar um cluster do Dataproc.

Para isso, você pode criar um cluster novo ou usar um cluster existente. Depois, use esse cluster para executar o Spark SQL interativo e gerenciar seus recursos de metastore do BigQuery.

  • A sub-rede na região em que o cluster é criado precisa ter o Acesso privado do Google (PGA, na sigla em inglês) ativado. Por padrão, as VMs do cluster do Dataproc, criadas com uma versão de imagem 2.2 (padrão) ou mais recente, têm apenas endereços IP internos. Para permitir que as VMs do cluster se comuniquem com as APIs do Google, ative o Acesso privado do Google na sub-rede de rede default (ou nome de rede especificado pelo usuário, se aplicável) na região em que o cluster foi criado.

  • Se você quiser executar o exemplo de interface da Web do Zeppelin neste guia, use ou crie um cluster do Dataproc com o componente opcional do Zeppelin ativado.

Novo cluster

Para criar um cluster do Dataproc, execute o comando gcloud dataproc clusters create a seguir. Essa configuração contém as configurações necessárias para usar a metastore do BigQuery.

gcloud dataproc clusters create CLUSTER_NAME \
  --project=PROJECT_ID \
  --region=LOCATION \
  --optional-components=ZEPPELIN \
  --enable-component-gateway \
  --single-node

Substitua:

  • CLUSTER_NAME: um nome para o cluster do Dataproc.
  • PROJECT_ID: o ID do projeto do Google Cloud em que você está criando o cluster.
  • LOCATION: a Google Cloud região em que você está criando o cluster.

Cluster existente

Para configurar um cluster, adicione o seguinte ambiente de execução do Iceberg Spark ao cluster.

org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.5.2

Você pode adicionar o ambiente de execução usando uma destas opções:

Enviar um job do Spark

Para enviar um job do Spark, use um destes métodos:

CLI da gcloud

gcloud dataproc jobs submit spark-sql \
--project=PROJECT_ID \
--cluster=CLUSTER_NAME \
--region==REGION \
--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 \
--properties=spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog, \
spark.sql.catalog.CATALOG_NAME.catalog-impl=org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog, \
spark.sql.catalog.CATALOG_NAME.gcp_project=PROJECT_ID, \
spark.sql.catalog.CATALOG_NAME.gcp_location=LOCATION, \
spark.sql.catalog.CATALOG_NAME.warehouse=WAREHOUSE_DIRECTORY \
--execute="SPARK_SQL_COMMAND"

Substitua:

  • PROJECT_ID: o ID do projeto do Google Cloud que contém o cluster do Dataproc.
  • CLUSTER_NAME: o nome do cluster do Dataproc que você está usando para executar o job do Spark SQL.
  • REGION: a região do Compute Engine em que o cluster está localizado.
  • LOCATION: o local dos recursos do BigQuery.
  • CATALOG_NAME: o nome do catálogo do Spark que você está usando com o job SQL.
  • WAREHOUSE_DIRECTORY: a pasta do Cloud Storage que contém o data warehouse. Esse valor começa com gs://.
  • SPARK_SQL_COMMAND: a consulta do Spark SQL que você quer executar. Essa consulta inclui os comandos para criar seus recursos. Por exemplo, para criar um namespace e uma tabela.

Faísca interativa

Conectar-se ao Spark e instalar o plug-in do catálogo

Para instalar o plug-in de catálogo para o metastore do BigQuery, conecte-se ao cluster do Dataproc usando SSH.

  1. No Console do Google Cloud, acesse a página Instâncias de VMs.

  2. Para se conectar a uma instância de VM do Dataproc, clique em SSH na lista de instâncias de máquina virtual. O resultado será assim:

    Connected, host fingerprint: ssh-rsa ...
Linux cluster-1-m 3.16.0-0.bpo.4-amd64 ...
...
example-cluster@cluster-1-m:~$

  3. No terminal, execute o seguinte comando de inicialização do metastore do BigQuery:

    spark-sql \
--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 \
--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

    Substitua:

    • CATALOG_NAME: o nome do catálogo do Spark que você está usando com o job SQL.
    • PROJECT_ID: o ID do projeto do Google Cloud do catálogo da metastore do BigQuery ao qual o catálogo do Spark está vinculado.
    • LOCATION: o Google Cloud local da metastore do BigQuery.
    • WAREHOUSE_DIRECTORY: a pasta do Cloud Storage que contém o data warehouse. Esse valor começa com gs://.

    Depois de se conectar a um cluster, o terminal do Spark vai mostrar a solicitação spark-sql.

    spark-sql (default)>

Gerenciar recursos do Metastore do BigQuery

Agora você está conectado à metastore do BigQuery. É possível visualizar os recursos atuais ou criar novos com base nos metadados armazenados na metastore do BigQuery.

Por exemplo, tente executar os comandos abaixo na sessão interativa do Spark SQL para criar um namespace e uma tabela do Iceberg.

  • Use o catálogo personalizado do Iceberg:

    USE `CATALOG_NAME`;

  • Para criar um namespace:

    CREATE NAMESPACE IF NOT EXISTS NAMESPACE_NAME;

  • Use o namespace criado:

    USE NAMESPACE_NAME;

  • Crie uma tabela Iceberg:

    CREATE TABLE TABLE_NAME (id int, data string) USING ICEBERG;

  • Insira uma linha de tabela:

    INSERT INTO TABLE_NAME VALUES (1, "first row");

  • Adicione uma coluna de tabela:

    ALTER TABLE TABLE_NAME ADD COLUMNS (newDoubleCol double);

  • Confira os metadados da tabela:

    DESCRIBE EXTENDED TABLE_NAME;

  • Liste as tabelas no namespace:

    SHOW TABLES;

Notebook Zeppelin

  1. No console do Google Cloud, acesse a página Clusters do Dataproc.

    Acessar clusters do Dataproc

  2. Clique no nome do cluster que você quer usar.

    A página Detalhes do cluster é aberta.

  3. No menu de navegação, clique em Interfaces da Web.

  4. Em Gateway de componentes, clique em Zeppelin. A página do notebook Zeppelin é aberta.

  5. No menu de navegação, clique em Notebook e em +Criar nova nota.

  6. Na caixa de diálogo, insira um nome para o notebook. Deixe Spark selecionado como o interpretador padrão.

  7. Clique em Criar. Um novo notebook é criado.

  8. No notebook, clique no menu de configurações e em Intérprete.

  9. No campo Pesquisar intérpretes, pesquise Spark.

  10. Clique em Editar.

  11. No campo Spark.jars, insira o URI do jar do Spark.

    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

  12. Clique em Salvar.

  13. Clique em OK.

  14. Copie o código PySpark abaixo no seu notebook do Zeppelin.

    %pyspark
from pyspark.sql import SparkSession
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.sql("select version()").show()
spark.sql("USE `CATALOG_NAME`;")
spark.sql("CREATE NAMESPACE IF NOT EXISTS NAMESPACE_NAME;")
spark.sql("USE NAMESPACE_NAME;")
spark.sql("CREATE TABLE TABLE_NAME (id int, data string) USING ICEBERG;")
spark.sql("DESCRIBE TABLE_NAME;").show()

    Substitua:

    • CATALOG_NAME: o nome do catálogo do Spark a ser usado para o job SQL.
    • PROJECT_ID: o ID do projeto do Google Cloud que contém o cluster do Dataproc.
    • WAREHOUSE_DIRECTORY: a pasta do Cloud Storage que contém o data warehouse. Esse valor começa com gs://.
    • NAMESPACE_NAME: o nome do namespace que faz referência à tabela do Spark.
    • WAREHOUSE_DIRECTORY: o URI da pasta do Cloud Storage em que o data warehouse é armazenado.
    • TABLE_NAME: um nome de tabela para a tabela do Spark.

  15. Clique no ícone de execução ou pressione Shift-Enter para executar o código. Quando o job é concluído, a mensagem de status mostra "Spark Job Finished", e a saída mostra o conteúdo da tabela:

A seguir