Configurar o Spark com tabelas Delta

Neste documento, explicamos como configurar o Apache Spark com o catálogo do ambiente de execução do Lakehouse para usar tabelas do Delta Lake. Você vai aprender a criar um catálogo do Apache Hive, configurar sessões do Spark para se conectar ao metastore e executar cargas de trabalho para criar e registrar tabelas Delta para consulta no BigQuery.

Antes de começar

  1. Leia Sobre o endpoint do catálogo do Hive.
  2. Analise os formatos de armazenamento e tipos de dados compatíveis.
  3. Analise as limitações e considerações.
  4. Faça login na sua Google Cloud conta do. Se você começou a usar o Google Cloud, crie uma conta para avaliar o desempenho dos nossos produtos em situações reais. Clientes novos também recebem US $300 em créditos para executar, testar e implantar cargas de trabalho.

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

  6. Enable the Lakehouse, Managed Service for Apache Spark APIs, if any are not already enabled.

    Roles required to enable APIs

    To enable APIs, you need the serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

    Enable the APIs

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

  8. Enable the Lakehouse, Managed Service for Apache Spark APIs, if any are not already enabled.

    Roles required to enable APIs

    To enable APIs, you need the serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

    Enable the APIs

    9.

Funções exigidas

Para ter as permissões necessárias para usar o catálogo do ambiente de execução do Lakehouse com tabelas Delta, peça ao administrador para conceder a você os seguintes papéis do IAM 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 com papéis personalizados ou outros papéis predefinidos.

Para instruções, consulte Conceder um único papel.

Fluxo de trabalho geral

Para usar o catálogo do ambiente de execução do Lakehouse com tabelas Delta, siga este fluxo de trabalho:

  1. Criar um catálogo Hive do Lakehouse.
  2. Configure sua sessão do Spark com o cliente do metastore do Lakehouse e as dependências do Delta Lake.
  3. Crie e preencha tabelas Delta no Cloud Storage usando o Spark, que as registra no metastore.
  4. Consulte as tabelas Delta do Apache Spark (leitura e gravação) ou diretamente do BigQuery (somente leitura).

Criar um catálogo Hive do Lakehouse

Para usar o catálogo do ambiente de execução do Lakehouse com tabelas do Spark e do Delta, crie um catálogo do Hive.

Um catálogo Hive do Lakehouse é uma coleção de bancos de dados do Hive. Antes de executar jobs do Spark, crie um catálogo para registrá-lo no metastore do Lakehouse. O catálogo exige um nome e um local do Cloud Storage para os dados.

Console

  1. No Google Cloud console do, abra a página Lakehouse.

    Acessar o Lakehouse

  2. Clique em Criar catálogo.

  3. Selecione catálogo do ambiente de execução do Lakehouse.

  4. Em Tipo de catálogo, selecione Metastore do Hive.

  5. No campo Selecionar um bucket do Cloud Storage, insira o nome do bucket do Cloud Storage para seu catálogo. Como alternativa, clique em Procurar para escolher ou criar um bucket.

  6. Em ID do catálogo, insira um nome para o catálogo Hive do Lakehouse.

  7. Em Local principal, especifique a mesma região do bucket.

  8. Clique em Criar.

gcloud

Para criar um catálogo do Hive, execute o seguinte comando:

gcloud biglake hive catalogs create LAKEHOUSE_CATALOG_ID \
    --project=PROJECT_ID \
    --location-uri="gs://GCS_WAREHOUSE_PATH" \
    --primary-location=REGION \
    --description="DESCRIPTION"

Substitua:

  • LAKEHOUSE_CATALOG_ID: o nome do catálogo do Hive.

  • GCS_WAREHOUSE_PATH: o caminho do Cloud Storage para os dados do data warehouse.

  • PROJECT_ID: o Google Cloud ID do projeto.

  • REGION: a região principal do metastore. Para buckets de região única, isso corresponde à região do bucket. Para buckets de região dupla ou multirregional, especifique uma região constituinte para a réplica principal. A outra região se torna a réplica secundária.

  • DESCRIPTION: uma descrição do catálogo.

Configurar o Spark para usar tabelas Delta

Para usar o catálogo de ambientes de execução do Lakehouse com tabelas Delta, configure sua sessão do Spark com a fábrica de clientes do metastore do Lakehouse e as extensões do Delta Lake.

spark-sql

  1. Use SSH para se conectar ao nó principal do cluster do Serviço Gerenciado para Apache Spark.

  2. Execute spark-sql na linha de comando com as seguintes propriedades para iniciar uma sessão interativa do Spark SQL:

    spark-sql \
    --conf spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension \
    --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog \
    --conf spark.hive.metastore.client.factory.class=com.google.cloud.bigquery.metastore.client.BigLakeMetastoreClientFactory \
    --conf spark.hive.metastore.blms.project.id=PROJECT_ID \
    --conf spark.hive.metastore.blms.catalog.default=LAKEHOUSE_CATALOG_ID \
    --conf spark.hive.metastore.warehouse.dir=gs://GCS_WAREHOUSE_PATH

  3. Depois que a sessão for iniciada, o Spark se conectará ao catálogo do ambiente de execução do Lakehouse.

    Execute os comandos a seguir para criar e consultar recursos Delta:

    -- Show all the databases in the current project.
SHOW DATABASES;

-- Create a database.
CREATE DATABASE spark_delta_blms_database;

-- Create a Delta data source table.
CREATE TABLE spark_delta_blms_database.delta_quick_start (id INT, name STRING) USING DELTA;

-- Insert data into the table.
INSERT INTO TABLE spark_delta_blms_database.delta_quick_start VALUES (1, 'my-first-user');

-- Query the Delta table.
SELECT * FROM spark_delta_blms_database.delta_quick_start;

    Substitua:

    • PROJECT_ID: o Google Cloud ID do projeto.

    • LAKEHOUSE_CATALOG_ID: o nome do catálogo do Hive.

    • GCS_WAREHOUSE_PATH: o caminho do Cloud Storage que armazena os dados do data warehouse.

Jupyter Notebook

  1. Execute um notebook Jupyter em um cluster do Serviço Gerenciado para Apache Spark.
  2. Acesse a interface da Web do Jupyter na guia Interfaces da Web da página de detalhes do cluster no Google Cloud console do.

  3. Em um novo notebook, crie uma sessão do Spark configurada para o Delta Lake e execute consultas:

    from pyspark.sql import SparkSession

# If a Spark session exists, stop it first by running spark.stop()
spark = SparkSession.builder\
    .master("local")\
    .appName("Lakehouse runtime catalog Delta tutorial")\
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")\
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")\
    .config("spark.hive.metastore.client.factory.class", "com.google.cloud.bigquery.metastore.client.BigLakeMetastoreClientFactory")\
    .config("spark.hive.metastore.blms.project.id", "PROJECT_ID")\
    .config("spark.hive.metastore.warehouse.dir", "gs://GCS_WAREHOUSE_PATH")\
    .config("spark.hive.metastore.blms.catalog.default", "LAKEHOUSE_CATALOG_ID")\
    .getOrCreate()

# Show all the databases.
df = spark.sql("SHOW DATABASES;")
df.show()

# Create a database.
spark.sql("CREATE DATABASE jupyter_delta_blms_db")

# Create a Delta data source table.
spark.sql("CREATE TABLE jupyter_delta_blms_db.delta_table(id INT, name STRING) USING DELTA")

# Insert data into the table.
spark.sql("INSERT INTO TABLE jupyter_delta_blms_db.delta_table VALUES (1, 'my-first-user');")

# Query from table.
spark.sql("SELECT * FROM jupyter_delta_blms_db.delta_table;").show()

    Substitua:

    • PROJECT_ID: o Google Cloud ID do projeto.

    • GCS_WAREHOUSE_PATH: o caminho do Cloud Storage que armazena os dados do data warehouse.

    • LAKEHOUSE_CATALOG_ID: o nome do catálogo do Hive.

Notebook do BigQuery

  1. No Google Cloud console, acesse o BigQuery.

    Acessar o BigQuery

  2. No painel Explorer , clique em + ADICIONAR e em Notebook Python.

  3. Em uma célula de código, configure a sessão do Serviço Gerenciado para Apache Spark com extensões Delta e propriedades do catálogo do ambiente de execução do Lakehouse:

    from google.cloud.dataproc_spark_connect import DataprocSparkSession
from google.cloud.dataproc_v1 import Session
import os

os.environ['DATAPROC_SPARK_CONNECT_DEFAULT_DATASOURCE'] = ""

session = Session()
session.environment_config.execution_config.ttl = {"seconds": 864000}
session.runtime_config.version = "2.3"
session.runtime_config.properties = {
 "spark.sql.extensions": "io.delta.sql.DeltaSparkSessionExtension",
 "spark.sql.catalog.spark_catalog": "org.apache.spark.sql.delta.catalog.DeltaCatalog",
 "spark.hive.metastore.blms.project.id": "PROJECT_ID",
 "spark.hive.metastore.blms.catalog.default": "LAKEHOUSE_CATALOG_ID",
 "spark.hive.metastore.warehouse.dir": "gs://GCS_WAREHOUSE_PATH",
 "spark.hive.metastore.client.factory.class": "com.google.cloud.bigquery.metastore.client.BigLakeMetastoreClientFactory",
 "spark.sql.catalogImplementation": "hive"
}

spark = DataprocSparkSession.builder.dataprocSessionConfig(session).getOrCreate()
print("Spark session created successfully")

  4. Em outra célula de código, crie um banco de dados e uma tabela Delta:

    # Create a database
spark.sql("CREATE DATABASE bq_spark_delta_database;")

# Create a Delta data source table
spark.sql("CREATE TABLE bq_spark_delta_database.delta_quick_start (id INT, name STRING) USING DELTA;")

# Insert data into the table
spark.sql("INSERT INTO TABLE bq_spark_delta_database.delta_quick_start VALUES (1, 'my-first-user');")

# Query from table
spark.sql("SELECT * FROM bq_spark_delta_database.delta_quick_start;").show()

    Substitua:

    • PROJECT_ID: o Google Cloud ID do projeto.

    • GCS_WAREHOUSE_PATH: o caminho do Cloud Storage que armazena os dados do data warehouse.

    • LAKEHOUSE_CATALOG_ID: o nome do catálogo do Hive.

Enviar um job em lote do Serviço Gerenciado para Apache Spark

É possível enviar uma carga de trabalho em lote do PySpark para o Serviço Gerenciado para Apache Spark para criar e consultar tabelas Delta com o catálogo do ambiente de execução do Lakehouse.

  1. Crie um arquivo Python com um job do PySpark:

    from pyspark.sql import SparkSession

spark = (
    SparkSession.builder.appName("Lakehouse runtime catalog Delta tutorial")
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
    .config(
        "spark.hive.metastore.client.factory.class",
        "com.google.cloud.bigquery.metastore.client.BigLakeMetastoreClientFactory",
    )
    .config("spark.hive.metastore.blms.project.id", "PROJECT_ID")
    .config("spark.hive.metastore.blms.catalog.default", "LAKEHOUSE_CATALOG_ID")
    .config(
        "spark.hive.metastore.warehouse.dir",
        "gs://GCS_WAREHOUSE_PATH",
    )
    .enableHiveSupport()
    .getOrCreate()
)

# Show all the databases.
spark.sql("SHOW DATABASES;").show()

# Create a database.
spark.sql("CREATE DATABASE dp_serverless_delta_db")

# Create a Delta data source table.
spark.sql(
    "CREATE TABLE dp_serverless_delta_db.delta_table(id INT, name STRING) USING DELTA"
)

# Insert data into the table.
spark.sql("INSERT INTO TABLE dp_serverless_delta_db.delta_table VALUES (1, 'my-first-user');")

# Query from table.
spark.sql("SELECT * FROM dp_serverless_delta_db.delta_table;").show()

    Substitua:

    • PROJECT_ID: o ID do Google Cloud projeto.

    • LAKEHOUSE_CATALOG_ID: o nome do catálogo do Hive.

    • GCS_WAREHOUSE_PATH: o caminho do Cloud Storage que armazena os dados do data warehouse.

  2. Envie o job em lote:

    gcloud dataproc batches submit pyspark PYTHON_SCRIPT_FILE \
    --version=2.2 \
    --project=PROJECT_ID \
    --region=REGION \
    --deps-bucket=gs://CLOUD_STORAGE_BUCKET

    Substitua:

    • PYTHON_SCRIPT_FILE: o caminho local ou do Cloud Storage para o arquivo do aplicativo PySpark.
    • PROJECT_ID: o ID do projeto.
    • REGION: a região em que o job em lote será executado.
    • CLOUD_STORAGE_BUCKET: o bucket do Cloud Storage para preparar as dependências da carga de trabalho.

Consultar tabelas Delta do BigQuery

Depois de criar tabelas Delta do Spark no catálogo do ambiente de execução do Lakehouse, é possível consultá-las no BigQuery Studio.

As consultas executadas no BigQuery em tabelas Delta no endpoint do catálogo do Hive são somente leitura.

  1. No Google Cloud console, acesse o BigQuery.

    Acessar o BigQuery

  2. No editor de consultas, digite a seguinte instrução:

    SELECT * FROM `PROJECT_ID.LAKEHOUSE_CATALOG_ID.DATABASE_NAME.TABLE_NAME`;

    Substitua:

    • PROJECT_ID: o ID do projeto.
    • LAKEHOUSE_CATALOG_ID: o nome do catálogo do Hive.
    • DATABASE_NAME: o nome do banco de dados.
    • TABLE_NAME: o nome da tabela Delta.

Limitações

Ao usar tabelas Delta com o endpoint do catálogo do Hive do catálogo do ambiente de execução do Lakehouse, as seguintes limitações se aplicam:

  • Consultas somente leitura no BigQuery: as consultas em tabelas Delta do BigQuery são somente leitura. Faça modificações de dados, operações DML e mudanças de esquema usando o Apache Spark.
  • Esquema de marcador de posição em GetHiveTable e inspeção de esquema verdadeiro: GetHiveTable chamadas para tabelas Delta retornam um esquema de marcador de posição em vez de o esquema de tabela real. Para inspecionar o esquema real de uma tabela Delta, use o Apache Spark. O BigQuery calcula um esquema modificado adaptado para a execução de consultas do Dremel, que não é uma correspondência 1:1 com o esquema real da tabela Delta.
  • Incompatibilidade do cliente do Hive: os clientes do Hive (como a CLI do Hive) não oferecem suporte a tabelas Delta com o endpoint do Hive do catálogo do ambiente de execução do Lakehouse. Use o Apache Spark para gerenciar tabelas Delta e o Spark ou o BigQuery para consultá-las.
  • Operações gerenciadas pelo catálogo não são compatíveis: o modo gerenciado pelo catálogo não é compatível com tabelas Delta. Não é possível criar uma tabela Delta usando o DDL do metastore independente. Em vez disso, crie tabelas Delta com locais de armazenamento externos no Cloud Storage (por exemplo, usando o Apache Spark) e registre-as no catálogo.
  • Limitações de tabelas Delta externas do BigQuery: o endpoint do Hive do catálogo do ambiente de execução do Lakehouse usa a mesma implementação de mecanismo de consulta subjacente que as tabelas Delta Lake externas do BigQuery. Todas as limitações de tabelas Delta Lake externas e limitações de tabelas BigLake se aplicam. Por exemplo, as operações de captura de dados alterados (CDC) e a API BigQuery Storage Read não são compatíveis, e timestamp_ntz não é compatível.

A seguir