Usar o metastore do BigLake com o Dataproc

Este documento explica como usar a metastore do BigLake 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 Apache Spark ou Apache Flink.

Antes de começar

1. Ative o faturamento do projeto 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 o BigLake Metastore funciona e por que você deve usá-lo.

Funções exigidas

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

• Crie um cluster do Dataproc: Worker do Dataproc (roles/dataproc.worker) na conta de serviço padrão do Compute Engine no projeto
• Crie tabelas do metastore do BigLake no Spark ou no Flink:
  • Worker do Dataproc (roles/dataproc.worker) na conta de serviço da VM do Dataproc no projeto
  • Editor de dados do BigQuery (roles/bigquery.dataEditor) na conta de serviço da VM do Dataproc no projeto
  • Administrador de objetos do Storage (roles/storage.objectAdmin) na conta de serviço da VM do Dataproc no projeto
• Consultar tabelas do metastore do BigLake no BigQuery:
  • Visualizador de dados do BigQuery (roles/bigquery.dataViewer) no projeto
  • Usuário do BigQuery (roles/bigquery.user) no projeto
  • Leitor de objetos do Storage (roles/storage.objectViewer) no projeto

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 BigLake, siga estas etapas gerais:

1. Crie ou configure um cluster do Dataproc.
2. Conecte-se ao mecanismo de software de código aberto preferido, como Spark ou Flink.
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 BigLake conforme necessário, dependendo do mecanismo de software de código aberto que você está usando.
5. No BigQuery, acesse e use os recursos do BigLake Metastore.

Conectar o BigLake Metastore ao Spark

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

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

Para conectar o BigLake Metastore ao Dataproc e ao Spark, use o arquivo jar do plug-in do catálogo Iceberg do Metastore do BigLake.

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 do Iceberg da metastore do BigLake.

Configurar um cluster do Dataproc

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

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

• 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.

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

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

Enviar um job do Spark

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

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.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 \
--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"

Conectar o BigLake Metastore ao Flink

As instruções a seguir mostram como conectar o Dataproc ao metastore do BigLake usando o cliente SQL do Flink.

Instalar o plug-in do catálogo e se conectar a uma sessão do Flink

Para conectar o BigLake Metastore ao Flink, faça o seguinte:

1. Crie um cluster do Dataproc com o componente Flink opcional ativado e verifique se você está usando o Dataproc 2.2 ou mais recente.

2. No Google Cloud console, acesse a página Instâncias de VM:

   Acessar instâncias de VM

3. Na lista de instâncias de máquina virtual, clique em SSH para se conectar a uma instância de VM do Dataproc.

4. Configure o plug-in do catálogo personalizado do Iceberg para o BigLake Metastore:

   FLINK_VERSION=1.17
   ICEBERG_VERSION=1.5.2
   
   cd /usr/lib/flink
   
   sudo wget -c https://repo.maven.apache.org/maven2/org/apache/iceberg/iceberg-flink-runtime-${FLINK_VERSION}/${ICEBERG_VERSION}/iceberg-flink-runtime-${FLINK_VERSION}-${ICEBERG_VERSION}.jar -P lib
   
   sudo gcloud storage cp gs://spark-lib/bigquery/iceberg-bigquery-catalog-${ICEBERG_VERSION}-1.0.1-beta.jar lib/

5. Inicie a sessão do Flink no YARN:

   HADOOP_CLASSPATH=`hadoop classpath`
   
   sudo bin/yarn-session.sh -nm flink-dataproc -d
   
   sudo bin/sql-client.sh embedded \
     -s yarn-session

6. Crie um catálogo no Flink:

   CREATE CATALOG CATALOG_NAME WITH (
     'type'='iceberg',
     'warehouse'='WAREHOUSE_DIRECTORY',
     'catalog-impl'='org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog',
     'gcp_project'='PROJECT_ID',
     'gcp_location'='LOCATION'
   );

   Substitua:

   • CATALOG_NAME: o identificador do catálogo do Flink, que é vinculado a um catálogo do BigLake Metastore.
   • WAREHOUSE_DIRECTORY: o caminho de base para o diretório do repositório (a pasta do Cloud Storage em que o Flink cria arquivos). Esse valor começa com gs://.
   • PROJECT_ID: o ID do projeto do catálogo do BigLake Metastore ao qual o catálogo do Flink está vinculado.
   • LOCATION: o local dos recursos do BigQuery.

Sua sessão do Flink agora está conectada à metastore do BigLake, e você pode executar comandos do Flink SQL.

Gerenciar recursos do Metastore do BigLake

Agora que você está conectado à metastore do BigLake, é possível criar e visualizar recursos com base nos metadados armazenados nela.

Por exemplo, tente executar os comandos abaixo na sua sessão interativa do Flink SQL para criar um banco de dados e uma tabela do Iceberg.

1. Use o catálogo personalizado do Iceberg:

   USE CATALOG CATALOG_NAME;

   Substitua CATALOG_NAME pelo identificador do catálogo do Flink.

2. Crie um banco de dados, que cria um conjunto de dados no BigQuery:

   CREATE DATABASE IF NOT EXISTS DATABASE_NAME;

   Substitua DATABASE_NAME pelo nome do novo banco de dados.

3. Use o banco de dados que você criou:

   USE DATABASE_NAME;

4. Crie uma tabela Iceberg. O exemplo a seguir cria uma tabela de vendas:

   CREATE TABLE IF NOT EXISTS ICEBERG_TABLE_NAME (
       order_number BIGINT,
       price        DECIMAL(32,2),
       buyer        ROW<first_name STRING, last_name STRING>,
       order_time   TIMESTAMP(3)
   );

   Substitua ICEBERG_TABLE_NAME por um nome para a nova tabela.

5. Acesse os metadados da tabela:

   DESCRIBE EXTENDED ICEBERG_TABLE_NAME;

6. Liste as tabelas no banco de dados:

   SHOW TABLES;

Ingerir dados na tabela

Depois de criar uma tabela Iceberg na seção anterior, use o Flink DataGen como uma fonte de dados para ingerir dados em tempo real na tabela. As etapas a seguir são um exemplo desse fluxo de trabalho:

1. Crie uma tabela temporária usando o DataGen:

   CREATE TEMPORARY TABLE DATABASE_NAME.TEMP_TABLE_NAME
   WITH (
       'connector' = 'datagen',
       'rows-per-second' = '10',
       'fields.order_number.kind' = 'sequence',
       'fields.order_number.start' = '1',
       'fields.order_number.end' = '1000000',
       'fields.price.min' = '0',
       'fields.price.max' = '10000',
       'fields.buyer.first_name.length' = '10',
       'fields.buyer.last_name.length' = '10'
   )
   LIKE DATABASE_NAME.ICEBERG_TABLE_NAME (EXCLUDING ALL);

   Substitua:

   • DATABASE_NAME: o nome do banco de dados para armazenar a tabela temporária.
   • TEMP_TABLE_NAME: um nome para a tabela temporária.
   • ICEBERG_TABLE_NAME: o nome da tabela Iceberg criada na seção anterior.

2. Defina o paralelismo como 1:

   SET 'parallelism.default' = '1';

3. Defina o intervalo de verificação:

   SET 'execution.checkpointing.interval' = '10second';

4. Defina o checkpoint:

   SET 'state.checkpoints.dir' = 'hdfs:///flink/checkpoints';

5. Inicie o job de streaming em tempo real:

   INSERT INTO ICEBERG_TABLE_NAME SELECT * FROM TEMP_TABLE_NAME;

   O resultado será assim:

   [INFO] Submitting SQL update statement to the cluster...
   [INFO] SQL update statement has been successfully submitted to the cluster:
   Job ID: 0de23327237ad8a811d37748acd9c10b
   

6. Para verificar o status do job de streaming, faça o seguinte:

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

      Acessar Clusters

   2. Selecione o cluster.

   3. Clique na guia Interfaces da Web.

   4. Clique no link YARN ResourceManager.

   5. Na interface do YARN ResourceManager, encontre sua sessão do Flink e clique no link ApplicationMaster em IU de rastreamento.

   6. Na coluna Status, confirme se o status do job é Em execução.

7. Consultar dados de streaming no cliente SQL do Flink:

   SELECT * FROM ICEBERG_TABLE_NAME
   /*+ OPTIONS('streaming'='true', 'monitor-interval'='3s')*/
   ORDER BY order_time desc
   LIMIT 20;

8. Consulte os dados de streaming no BigQuery:

   SELECT * FROM `DATABASE_NAME.ICEBERG_TABLE_NAME`
   ORDER BY order_time desc
   LIMIT 20;

9. Encerrar o job de streaming no cliente SQL do Flink:

   STOP JOB 'JOB_ID';

   Substitua JOB_ID pelo ID do job exibido na saída quando você criou o job de streaming.

A seguir

• Configure os recursos opcionais do metastore do BigLake.
• Acessar e consultar tabelas do Spark no BigQuery.