Usar o metastore do BigLake com o Dataproc

Neste documento, explicamos como usar o 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 o Apache Spark ou o Apache Flink.

Antes de começar

1. Ative o faturamento no 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 o metastore do BigLake como um repositório de metadados, 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
• Consulte as tabelas do metastore do BigLake no BigQuery:
  • Leitor 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 o metastore do BigLake, siga estas etapas gerais:

1. Crie ou configure um cluster do Dataproc.
2. Conecte-se ao seu 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 os 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 metastore do BigLake.

Conectar o BigLake Metastore ao Spark

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

Baixar o plug-in do catálogo do Iceberg

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

Esse arquivo está 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, baixe o 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 do metastore do BigLake.

Configurar um cluster do Dataproc

Antes de se conectar ao metastore do BigLake, configure um cluster do Dataproc.

Para isso, crie um cluster ou use um já existente. Depois, você usa esse cluster para executar o Spark SQL interativo e gerenciar os recursos do metastore do BigLake.

• A sub-rede na região em que o cluster é criado precisa ter o Acesso privado do Google (PGA) ativado. Por padrão, as VMs de cluster do Dataproc, criadas com uma versão de imagem 2.2 (padrão) ou posterior, 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 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 da 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 dos seguintes 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 metastore do BigLake ao Flink

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

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

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

1. Crie um cluster do Dataproc com o componente opcional Flink ativado e verifique se você está usando o Dataproc 2.2 ou uma versão 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 está vinculado a um catálogo do BigLake Metastore.
   • WAREHOUSE_DIRECTORY: o caminho base para o diretório do data warehouse (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 ao metastore do BigLake, e você pode executar comandos SQL do Flink.

Gerenciar recursos do metastore do BigLake

Agora que você está conectado ao BigLake Metastore, é possível criar e visualizar recursos com base nos metadados armazenados nele.

Por exemplo, execute os comandos a seguir 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 do Iceberg. O comando a seguir cria uma tabela de vendas de exemplo:

   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. Ver 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 do Iceberg na seção anterior, use o Flink DataGen como uma fonte de dados para ingerir dados em tempo real na sua 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 sua tabela temporária.
   • TEMP_TABLE_NAME: um nome para sua tabela temporária.
   • ICEBERG_TABLE_NAME: o nome da tabela do Iceberg que você criou na seção anterior.

2. Defina o paralelismo como 1:

   SET 'parallelism.default' = '1';

3. Defina o intervalo de checkpoint:

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

4. Defina o checkpoint:

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

5. Inicie o job de transmissão 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 console Google Cloud , 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. Consultar dados de streaming no BigQuery:

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

9. Encerre o job de streaming no cliente Flink SQL:

   STOP JOB 'JOB_ID';

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

A seguir

• Configure os recursos opcionais do metastore do BigLake.
• Consulte e visualize tabelas do Spark no BigQuery.