Configurar o metastore do BigLake
Neste documento, explicamos como configurar o metastore do BigLake com o Dataproc ou o Google Cloud sem servidor para Apache Spark e criar um metastore único e compartilhado que funcione em mecanismos de código aberto, como o Apache Spark ou o Apache Flink.
Antes de começar
- Ative o faturamento no projeto Google Cloud . Saiba como verificar se o faturamento está ativado em um projeto.
Ative as APIs BigQuery e Dataproc.
Opcional: entenda como a metastore do BigLake funciona e por que você deve usá-la.
Funções exigidas
Para receber as permissões necessárias para configurar o metastore do BigLake, 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:
-
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
-
Worker do Dataproc (
-
Consulte tabelas do metastore do BigLake:
-
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
-
Leitor de dados do BigQuery (
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 usando papéis personalizados ou outros papéis predefinidos.
Configurar o metastore com o Dataproc
É possível configurar o BigLake Metastore com o Dataproc usando o Spark ou o Flink:
Spark
Configure um novo cluster. Para criar um cluster do Dataproc, execute o seguinte comando
gcloud dataproc clusters create
, que contém as configurações necessárias para usar o metastore do BigLake:gcloud dataproc clusters create CLUSTER_NAME \ --project=PROJECT_ID \ --region=LOCATION \ --single-node
Substitua:
CLUSTER_NAME
: um nome para o cluster do Dataproc.PROJECT_ID
: o ID do Google Cloud projeto em que você está criando o cluster.LOCATION
: a região do Compute Engine em que você está criando o cluster.
Envie um job do Spark usando um dos seguintes métodos:
CLI do Google Cloud
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"
Substitua:
PROJECT_ID
: o ID do Google Cloud projeto 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 a ser usado com seu job do SQL.WAREHOUSE_DIRECTORY
: a pasta do Cloud Storage que contém seu data warehouse. Esse valor começa comgs://
.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.
CLI do spark-sql
No console do Google Cloud , acesse a página Instâncias de VM.
Para se conectar a uma instância de VM do Dataproc, clique em SSH na linha que lista o nome da instância de VM principal do cluster do Dataproc, que é o nome do cluster seguido por um sufixo
-m
. 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:~$
No terminal, execute o seguinte comando de inicialização do metastore do BigLake:
spark-sql \ --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 \ --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 seu job do SQL.PROJECT_ID
: o ID do Google Cloud projeto do catálogo do BigLake Metastore ao qual seu catálogo do Spark está vinculado.LOCATION
: o Google Cloud local do metastore do BigLake.WAREHOUSE_DIRECTORY
: a pasta do Cloud Storage que contém seu data warehouse. Esse valor começa comgs://
.
Depois de se conectar ao cluster, o terminal do Spark vai mostrar o prompt
spark-sql
, que pode ser usado para enviar jobs do Spark.spark-sql (default)>
Flink
- 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. No console do Google Cloud , acesse a página Instâncias de VM.
Na lista de instâncias de máquina virtual, clique em SSH para se conectar à instância principal de VM do cluster do Dataproc, que aparece como o nome do cluster seguido por um sufixo
-m
.Configure o plug-in do catálogo personalizado do Iceberg para o metastore do BigLake:
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/
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
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 comgs://
.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 BigLake Metastore, e você pode executar comandos SQL do Flink.
Gerenciar recursos do BigLake Metastore
Agora que você está conectado ao BigLake Metastore, é possível criar e visualizar recursos com base nos metadados armazenados nele.
Por exemplo, execute os seguintes comandos na sua sessão interativa do Flink SQL para criar um banco de dados e uma tabela do Iceberg.
Use o catálogo personalizado do Iceberg:
USE CATALOG CATALOG_NAME;
Substitua
CATALOG_NAME
pelo identificador do catálogo do Flink.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.Use o banco de dados que você criou:
USE DATABASE_NAME;
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.Ver metadados da tabela:
DESCRIBE EXTENDED ICEBERG_TABLE_NAME;
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:
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.
Defina o paralelismo como 1:
SET 'parallelism.default' = '1';
Defina o intervalo de checkpoint:
SET 'execution.checkpointing.interval' = '10second';
Defina o checkpoint:
SET 'state.checkpoints.dir' = 'hdfs:///flink/checkpoints';
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
Para verificar o status do job de streaming, faça o seguinte:
No Google Cloud console, acesse a página Clusters.
Selecione o cluster.
Clique na guia Interfaces da Web.
Clique no link YARN ResourceManager.
Na interface do YARN ResourceManager, encontre sua sessão do Flink e clique no link ApplicationMaster em IU de rastreamento.
Na coluna Status, confirme se o status do job é Em execução.
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;
Consultar dados de streaming no BigQuery:
SELECT * FROM `DATABASE_NAME.ICEBERG_TABLE_NAME` ORDER BY order_time desc LIMIT 20;
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.
Configurar o metastore com o Serverless para Apache Spark
É possível configurar o metastore do BigLake com o Serverless para Apache Spark usando o Spark SQL ou o PySpark.
Spark SQL
Crie um arquivo SQL com os comandos do Spark SQL que você quer executar no metastore do BigLake. Por exemplo, este comando cria um namespace e uma tabela:
CREATE NAMESPACE `CATALOG_NAME`.NAMESPACE_NAME; CREATE TABLE `CATALOG_NAME`.NAMESPACE_NAME.TABLE_NAME (id int, data string) USING ICEBERG LOCATION 'WAREHOUSE_DIRECTORY';
Substitua:
CATALOG_NAME
: o nome do catálogo que faz referência à sua tabela do Spark.NAMESPACE_NAME
: o nome do namespace que referencia sua tabela do Spark.TABLE_NAME
: um nome para sua tabela do Spark.WAREHOUSE_DIRECTORY
: o URI da pasta do Cloud Storage em que seu data warehouse está armazenado.
Envie um job em lote do Spark SQL executando o seguinte comando
gcloud dataproc batches submit spark-sql
:gcloud dataproc batches submit spark-sql SQL_SCRIPT_PATH \ --project=PROJECT_ID \ --region=REGION \ --subnet=projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_NAME \ --deps-bucket=BUCKET_PATH \ --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, \ .sql.catalog.CATALOG_NAME.warehouse=WAREHOUSE_DIRECTORY"
Substitua:
SQL_SCRIPT_PATH
: o caminho para o arquivo SQL que o job em lote usa.PROJECT_ID
: o ID do Google Cloud projeto em que o job em lote será executado.REGION
: a região em que sua carga de trabalho é executada.SUBNET_NAME
(opcional): o nome de uma sub-rede VPC noREGION
que atende aos requisitos de sub-rede da sessão.BUCKET_PATH
: o local do bucket do Cloud Storage para fazer upload das dependências da carga de trabalho. OWAREHOUSE_DIRECTORY
está localizado neste bucket. O prefixo de URIgs://
do bucket não é necessário. É possível especificar o caminho ou o nome do bucket, por exemplo,mybucketname1
.LOCATION
: o local para executar o job em lote.
Para mais informações sobre como enviar jobs em lote do Spark, consulte Executar uma carga de trabalho em lote do Spark.
PySpark
Crie um arquivo Python com os comandos do PySpark que você quer executar no BigLake Metastore.
Por exemplo, o comando a seguir configura um ambiente do Spark para interagir com tabelas do Iceberg armazenadas no metastore do BigLake. Em seguida, o comando cria um novo namespace e uma tabela do Iceberg dentro dele.
from pyspark.sql import SparkSession spark = SparkSession.builder \ .appName("BigLake 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("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 LOCATION 'WAREHOUSE_DIRECTORY';")
Substitua:
PROJECT_ID
: o ID do Google Cloud projeto em que o job em lote será executado.LOCATION
: o local em que os recursos do BigQuery estão localizados.CATALOG_NAME
: o nome do catálogo que faz referência à sua tabela do Spark.TABLE_NAME
: um nome para sua tabela do Spark.WAREHOUSE_DIRECTORY
: o URI da pasta do Cloud Storage em que seu data warehouse está armazenado.NAMESPACE_NAME
: o nome do namespace que referencia sua tabela do Spark.
Envie o job em lote usando o seguinte comando
gcloud dataproc batches submit pyspark
:gcloud dataproc batches submit pyspark PYTHON_SCRIPT_PATH \ --version=2.2 \ --project=PROJECT_ID \ --region=REGION \ --deps-bucket=BUCKET_PATH \ --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"
Substitua:
PYTHON_SCRIPT_PATH
: o caminho para o script Python usado pelo job em lote.PROJECT_ID
: o ID do Google Cloud projeto em que o job em lote será executado.REGION
: a região em que sua carga de trabalho é executada.BUCKET_PATH
: o local do bucket do Cloud Storage para fazer upload das dependências da carga de trabalho. O prefixo de URIgs://
do bucket não é necessário. É possível especificar o caminho ou o nome do bucket, por exemplo,mybucketname1
.
Para mais informações sobre como enviar jobs em lote do PySpark, consulte a referência da gcloud do PySpark.