Consulta catálogos federados con Spark y el catálogo de tiempo de ejecución de Lakehouse

Vista previa

Esta función está sujeta a las "Condiciones de la oferta previa a la DG" en la sección Condiciones Generales del Servicio de las Condiciones Específicas del Servicio. Las funciones de la fase previa a la DG están disponibles “sin modificaciones” y pueden tener asistencia limitada. Para obtener más información, consulta las descripciones de la etapa de lanzamiento.

Después de configurar el acceso a los datos multinube en el catálogo de tiempo de ejecución de Lakehouse, puedes consultar tablas de Iceberg remotas desde Managed Service para Apache Spark sin migrar archivos ni administrar credenciales de almacenamiento en la nube independientes.

El catálogo de tiempo de ejecución de Lakehouse controla la sincronización de metadatos y el acceso seguro a los datos a través de la venta de credenciales. Cuando Spark consulta una tabla federada, el catálogo de tiempo de ejecución de Lakehouse traduce los metadatos del catálogo y emite credenciales de corta duración y con alcance para que Spark lea datos de Amazon S3 o Cloud Storage, lo que trata las tablas como si fueran locales para tu entorno deGoogle Cloud .

Tipos de catálogos federados admitidos

Managed Service para Apache Spark admite la consulta de los siguientes tipos de catálogos federados a través del extremo del catálogo de REST de Iceberg del catálogo de tiempo de ejecución de Lakehouse:

Tipo de catálogo federado	Backends de almacenamiento compatibles	Guía de configuración en el catálogo de tiempo de ejecución de Lakehouse
Catálogo de datos de AWS Glue	Amazon S3 (s3://: catálogo de cuentas de AWS predeterminado o bucket de tabla de Amazon S3)	Conéctate a AWS Glue
Databricks Unity Catalog	Ubicaciones externas de Amazon S3 (s3://) o Cloud Storage (gs://)	Conéctate a Databricks Unity Catalog
Catálogo de Snowflake Horizon	Amazon S3 (s3://) o Cloud Storage (gs://)	Conéctate al catálogo de Snowflake Horizon

Para obtener información sobre los requisitos de preparación del catálogo y las definiciones de los parámetros de Spark subyacentes para cada proveedor de almacenamiento, consulta Consulta datos remotos en el catálogo de tiempo de ejecución de Lakehouse y Parámetros de configuración de Spark.

Antes de comenzar

Accede a tu cuenta de Google Cloud . Si eres nuevo en Google Cloud, crea una cuenta para evaluar el rendimiento de nuestros productos en situaciones reales. Los clientes nuevos también obtienen $300 en créditos gratuitos para ejecutar, probar y, además, implementar cargas de trabajo.

In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

Roles required to select or create a project

• Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
• Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

Go to project selector

Verify that you have the permissions required to complete this guide.

Verify that billing is enabled for your Google Cloud project.

Enable the Dataproc, BigLake, and BigQuery 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

In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

Roles required to select or create a project

• Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
• Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

Go to project selector

Verify that you have the permissions required to complete this guide.

Verify that billing is enabled for your Google Cloud project.

Enable the Dataproc, BigLake, and BigQuery 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

1. Asegúrate de tener tablas de Iceberg en tu catálogo remoto.
2. Configura una conexión multinube y un catálogo federado en el catálogo de tiempo de ejecución de Lakehouse para tu proveedor de catálogo remoto:
   • Catálogo de datos de AWS Glue
   • Databricks Unity Catalog
   • Catálogo de Snowflake Horizon

Roles de Identity and Access Management (IAM) necesarios

Se requieren ciertos roles de IAM para ejecutar los ejemplos de esta página. Según las políticas de la organización, es posible que ya se hayan otorgado estos roles. Para verificar las asignaciones de roles, consulta ¿Necesitas otorgar roles?. Para obtener más información para otorgar roles, consulta Administra el acceso a proyectos, carpetas y organizaciones.

Los roles que necesitas dependen de cómo se ejecuta tu carga de trabajo. Los clústeres de Managed Service para Apache Spark y los entornos de ejecución sin servidores de Managed Service para Apache Spark 2.3 y versiones anteriores ejecutan cargas de trabajo como una cuenta de servicio. Los entornos de ejecución sin servidores 3.0 y posteriores de Managed Service para Apache Spark ejecutan cargas de trabajo con tus credenciales de usuario final (EUC) de forma predeterminada, por lo que otorgas roles de acceso a los datos a tu cuenta de usuario en lugar de a una cuenta de servicio. Si especificas una cuenta de servicio personalizada para un entorno de ejecución 3.0 o posterior, la carga de trabajo se ejecuta con las credenciales de la cuenta de servicio. Para obtener más información, consulta Roles de implementaciones sin servidores.

Funciones de usuario

Para obtener los permisos que necesitas para crear un clúster de Managed Service para Apache Spark, enviar una carga de trabajo por lotes o crear una sesión interactiva, pídele a tu administrador que te otorgue los siguientes roles de IAM:

• Usa un clúster o un entorno de ejecución sin servidores 2.3 o anterior:
  • Editor de Dataproc (roles/dataproc.editor) en el proyecto
  • Usuario de cuenta de servicio (roles/iam.serviceAccountUser) en la cuenta de servicio predeterminada de Compute Engine o en una cuenta de servicio personalizada
• Usa un entorno de ejecución sin servidores 3.0 o posterior: Editor de Dataproc Serverless (roles/dataproc.serverlessEditor) en el proyecto
• Lee tablas federadas con credenciales de usuario final en un entorno de ejecución sin servidores 3.0 o posterior: Visualizador de BigLake (roles/biglake.viewer) en el proyecto que contiene el catálogo federado
• Usa las credenciales de cuenta de servicio en un entorno de ejecución sin servidores 3.0 o posterior: Usuario de la cuenta de servicio (roles/iam.serviceAccountUser) en la cuenta de servicio personalizada

Roles de la cuenta de servicio

Otorga los siguientes roles a la cuenta de servicio que ejecuta tu carga de trabajo. Los entornos de ejecución sin servidores 3.0 y posteriores de Managed Service para Apache Spark necesitan estos roles solo si la carga de trabajo se ejecuta con credenciales de cuenta de servicio.

Para garantizar que la cuenta de servicio predeterminada de Compute Engine o una cuenta de servicio personalizada tengan los permisos necesarios para ejecutar trabajos de Managed Service para Apache Spark y leer tablas federadas, pídele a tu administrador que otorgue los siguientes roles de IAM a la cuenta de servicio predeterminada de Compute Engine o a una cuenta de servicio personalizada:

Importante: Debes otorgar estos roles a la cuenta de servicio predeterminada de Compute Engine o a una cuenta de servicio personalizada, no a tu cuenta de usuario. Si no se otorgan los roles a la principal correcta, es posible que se produzcan errores de permisos.

• Ejecuta trabajos de Managed Service para Apache Spark en un clúster o un entorno de ejecución sin servidores 2.3 o anterior: Trabajador de Dataproc (roles/dataproc.worker) en el proyecto
• Ejecuta trabajos de Managed Service para Apache Spark en un entorno de ejecución sin servidores 3.0 o posterior:
  • Nodo sin servidores de Dataproc (roles/dataproc.serverlessNode) en el proyecto
  • Agente de servicio del nodo de Dataproc Resource Manager (roles/dataprocrm.nodeServiceAgent) en el proyecto
• Descubre y lee metadatos de tablas federadas en el entorno de ejecución: Visualizador de BigLake (roles/biglake.viewer) en el proyecto que contiene el catálogo federado

Para garantizar que el agente de servicio de Managed Service para Apache Spark ([e-mail]) tenga los permisos necesarios para resolver la propiedad de configuración simplificada dataproc.lakehouse.catalog. cuando creas un clúster, envías una carga de trabajo por lotes o creas una sesión interactiva, pídele a tu administrador que otorgue el rol de IAM de visualizador de BigLake (roles/biglake.viewer) al agente de servicio de Managed Service para Apache Spark ([e-mail]) en el proyecto que contiene el catálogo federado.

Importante: Debes otorgar este rol al agente de servicio de Managed Service para Apache Spark ([e-mail]), no a tu cuenta de usuario. Si no se otorga el rol a la principal correcta, se pueden producir errores de permisos.

Configura la conectividad de red saliente para Amazon S3

Cuando consultas catálogos federados respaldados por Amazon S3, las cargas de trabajo por lotes sin servidores y las sesiones interactivas de Managed Service para Apache Spark, y los clústeres de Managed Service para Apache Spark con IP interna no pueden conectarse a los extremos de Amazon S3 con su configuración de red predeterminada. Aprovisiona un Cloud Router y Cloud NAT en tu red de VPC de carga de trabajo para permitir el tráfico saliente a Amazon S3:

gcloud compute routers create lakehouse-router \
    --network=NETWORK_NAME \
    --region=REGION

gcloud compute routers nats create lakehouse-nat \
    --router=lakehouse-router \
    --auto-allocate-nat-external-ips \
    --nat-all-subnet-ip-ranges \
    --region=REGION

Reemplaza lo siguiente:

• NETWORK_NAME: Es la red de VPC para tu clúster de Managed Service para Apache Spark, la carga de trabajo por lotes o la sesión interactiva, por ejemplo, default.
• REGION: Es la región de Google Cloud para tu clúster de Managed Service para Apache Spark, carga de trabajo por lotes o sesión interactiva.

Configura Managed Service para Apache Spark y consulta tablas federadas

Puedes configurar un clúster de Managed Service para Apache Spark, una carga de trabajo por lotes sin servidores o una sesión interactiva para conectarte a tu catálogo federado en el catálogo de tiempo de ejecución de Lakehouse. Estos recursos de Managed Service para Apache Spark admiten la propiedad de configuración simplificada dataproc.lakehouse.catalog.CATALOG_NAME (recomendada), que lee la definición del catálogo federado del catálogo de tiempo de ejecución de Lakehouse y genera las propiedades de Spark del catálogo de REST de Iceberg necesarias, incluidos los encabezados de almacenamiento io-impl y de venta de credenciales (el formato de clúster de la propiedad agrega el prefijo dataproc:).

Tanto la configuración simplificada como la manual usan la autorización de Google para Iceberg (GoogleAuthManager), que requiere una de las siguientes versiones de imagen de clúster o de entorno de ejecución sin servidores:

• Clústeres de Managed Service para Apache Spark: versiones de imagen 2.2, 2.2.65 y posteriores, o versiones de imagen 2.3, 2.3.11 y posteriores.
• Cargas de trabajo por lotes sin servidores y sesiones interactivas de Managed Service para Apache Spark: entornos de ejecución 2.2.60 y posteriores de la versión 2.2, entornos de ejecución 2.3.10 y posteriores de la versión 2.3, o entornos de ejecución de la versión 3.0.

Para las cargas de trabajo por lotes sin servidores y las sesiones interactivas, los roles requeridos dependen de la versión del entorno de ejecución. Los entornos de ejecución 3.0 y posteriores usan credenciales de usuario final de forma predeterminada. Para obtener más detalles, consulta Roles de IAM obligatorios y Roles de implementaciones sin servidores.

Clúster

Crea un clúster de Managed Service para Apache Spark con el componente opcional de Iceberg usando propiedades de configuración simplificadas (recomendado) o especificando manualmente las propiedades del catálogo.

Configuración simplificada con propiedades (recomendado)

Crea el clúster con la propiedad de catálogo federado:

gcloud dataproc clusters create CLUSTER_NAME \
  --project=PROJECT_ID \
  --region=REGION \
  --image-version=IMAGE_VERSION \
  --optional-components=ICEBERG,JUPYTER \
  --enable-component-gateway \
  --properties="dataproc:dataproc.lakehouse.catalog.CATALOG_NAME=projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME,spark:spark.sql.defaultCatalog=CATALOG_NAME"

Reemplaza lo siguiente:

• CLUSTER_NAME: Es un nombre para tu clúster.
• PROJECT_ID: Es el ID del proyecto deGoogle Cloud .
• REGION: Es la región de Google Cloud para el clúster, por ejemplo, us-central1.
• IMAGE_VERSION: Es la versión de la imagen de Managed Service para Apache Spark, por ejemplo, 2.3-debian12. Para conocer las versiones compatibles, consulta los requisitos que se mencionan más arriba en esta sección.
• CATALOG_NAME: Es un nombre para el catálogo local de Spark, por ejemplo, federated_catalog. Puede ser igual a la de FEDERATED_CATALOG_NAME.
• FEDERATED_CATALOG_NAME: Es el ID del catálogo federado que creaste en Antes de comenzar.

Nota: Puedes pasar varios catálogos separados por comas a la marca --properties. Cada propiedad del catálogo de clústeres requiere el prefijo dataproc:.

Consulta el catálogo de Snowflake Horizon

Si consultas un catálogo de Snowflake Horizon, ten en cuenta lo siguiente:

• Almacenamiento de Snowflake: La configuración simplificada admite tablas de Snowflake almacenadas en Amazon S3. Para las tablas de Snowflake almacenadas en Cloud Storage, usa la configuración manual con io-impl establecido en org.apache.iceberg.gcp.gcs.GCSFileIO.

• Columnas STRING de Snowflake: Si encuentras errores cuando consultas columnas STRING de Snowflake en las que la política de serialización de almacenamiento de la tabla en Snowflake no es COMPATIBLE, puedes cambiar la política de serialización de almacenamiento a COMPATIBLE en Snowflake o inhabilitar la vectorización de Parquet en Spark agregando las siguientes propiedades a la marca --properties:

  spark:spark.sql.iceberg.vectorization.enabled=false,spark:spark.sql.catalog.CATALOG_NAME.table-override.read.parquet.vectorization.enabled=false
      

En tu notebook de Jupyter o aplicación de PySpark, crea el SparkSession sin especificar las configuraciones del catálogo:

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("APP_NAME").getOrCreate()

Reemplaza APP_NAME por el nombre de tu aplicación de Spark.

Configuración manual

Si no usas la propiedad de configuración simplificada, crea el clúster como se describió antes en esta sección, sin pasar la propiedad dataproc:dataproc.lakehouse.catalog. a la marca --properties y, luego, configura tu SparkSession de forma manual. Para obtener una lista de los valores de los parámetros del proveedor de almacenamiento, consulta Parámetros de configuración de Spark.

from pyspark.sql import SparkSession

catalog_name = "CATALOG_NAME"

spark = SparkSession.builder.appName("APP_NAME") \
.config('spark.sql.defaultCatalog', catalog_name) \
.config(f'spark.sql.catalog.{catalog_name}', 'org.apache.iceberg.spark.SparkCatalog') \
.config(f'spark.sql.catalog.{catalog_name}.type', 'rest') \
.config(f'spark.sql.catalog.{catalog_name}.uri', 'https://biglake.googleapis.com/iceberg/v1/restcatalog') \
.config(f'spark.sql.catalog.{catalog_name}.warehouse', 'bl://projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME') \
.config(f'spark.sql.catalog.{catalog_name}.header.x-goog-user-project', 'PROJECT_ID') \
.config(f'spark.sql.catalog.{catalog_name}.rest.auth.type', 'org.apache.iceberg.gcp.auth.GoogleAuthManager') \
.config(f'spark.sql.catalog.{catalog_name}.header.X-Iceberg-Access-Delegation', 'vended-credentials') \
.config(f'spark.sql.catalog.{catalog_name}.io-impl', 'IO_IMPL') \
.config(f'spark.sql.catalog.{catalog_name}.s3.client-factory-impl', 'org.apache.iceberg.aws.s3.DefaultS3FileIOAwsClientFactory') \
.config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions') \
.getOrCreate()

Reemplaza IO_IMPL por la implementación de FileIO para el almacenamiento que contiene tus tablas:

• Tablas en Amazon S3: org.apache.iceberg.aws.s3.S3FileIO
• Tablas en Cloud Storage: org.apache.iceberg.gcp.gcs.GCSFileIO

La propiedad s3.client-factory-impl solo se aplica a las tablas de Amazon S3. Omite esta línea para las tablas de Cloud Storage.

Sin servidores

Envía una carga de trabajo por lotes de PySpark a Managed Service para Apache Spark sin servidores con propiedades de configuración simplificadas (recomendado) o especificando manualmente las propiedades del catálogo.

Configuración simplificada con propiedades (recomendado)

Pasa la propiedad del catálogo federado sin el prefijo dataproc: cuando envíes la carga de trabajo por lotes:

gcloud dataproc batches submit pyspark PYSPARK_FILE \
  --project=PROJECT_ID \
  --region=REGION \
  --version=RUNTIME_VERSION \
  --properties="dataproc.lakehouse.catalog.CATALOG_NAME=projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME,spark.sql.defaultCatalog=CATALOG_NAME"

Reemplaza lo siguiente:

• PYSPARK_FILE: Es la ruta de acceso de Cloud Storage de gs:// al archivo de tu aplicación de PySpark.
• PROJECT_ID: Es el ID del proyecto deGoogle Cloud .
• REGION: Es la región de la carga de trabajo por lotes.
• RUNTIME_VERSION: Es la versión del entorno de ejecución sin servidores de Managed Service para Apache Spark, por ejemplo, 3.0. Para conocer las versiones compatibles, consulta los requisitos que se mencionan más arriba en esta sección. Los roles de IAM necesarios dependen de la versión del entorno de ejecución.
• CATALOG_NAME: Es un nombre para el catálogo local de Spark, por ejemplo, federated_catalog. Puede ser igual a la de FEDERATED_CATALOG_NAME.
• FEDERATED_CATALOG_NAME: Es el ID del catálogo federado que creaste en Antes de comenzar.

Nota: Puedes pasar varios catálogos separados por comas a la marca --properties.

Consulta el catálogo de Snowflake Horizon

Si consultas un catálogo de Snowflake Horizon, ten en cuenta lo siguiente:

• Almacenamiento de Snowflake: La configuración simplificada admite tablas de Snowflake almacenadas en Amazon S3. Para las tablas de Snowflake almacenadas en Cloud Storage, usa la configuración manual con io-impl establecido en org.apache.iceberg.gcp.gcs.GCSFileIO.

• Columnas STRING de Snowflake: Si encuentras errores cuando consultas columnas STRING de Snowflake en las que la política de serialización de almacenamiento de la tabla en Snowflake no es COMPATIBLE, puedes cambiar la política de serialización de almacenamiento a COMPATIBLE en Snowflake o inhabilitar la vectorización de Parquet en Spark agregando las siguientes propiedades a la marca --properties:

  spark.sql.iceberg.vectorization.enabled=false,spark.sql.catalog.CATALOG_NAME.table-override.read.parquet.vectorization.enabled=false
      

En el archivo de la aplicación de PySpark, crea el SparkSession sin especificar las configuraciones del catálogo:

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("APP_NAME").getOrCreate()

Configuración manual

Si no usas la propiedad de configuración simplificada, especifica las propiedades del catálogo con la marca --properties del comando gcloud dataproc batches submit. Para obtener una lista de los valores de los parámetros del proveedor de almacenamiento, consulta Parámetros de configuración de Spark.

gcloud dataproc batches submit pyspark PYSPARK_FILE \
  --project=PROJECT_ID \
  --region=REGION \
  --version=RUNTIME_VERSION \
  --properties="\
spark.sql.defaultCatalog=CATALOG_NAME,\
spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog,\
spark.sql.catalog.CATALOG_NAME.type=rest,\
spark.sql.catalog.CATALOG_NAME.uri=https://biglake.googleapis.com/iceberg/v1/restcatalog,\
spark.sql.catalog.CATALOG_NAME.warehouse=bl://projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME,\
spark.sql.catalog.CATALOG_NAME.header.x-goog-user-project=PROJECT_ID,\
spark.sql.catalog.CATALOG_NAME.rest.auth.type=org.apache.iceberg.gcp.auth.GoogleAuthManager,\
spark.sql.catalog.CATALOG_NAME.header.X-Iceberg-Access-Delegation=vended-credentials,\
spark.sql.catalog.CATALOG_NAME.io-impl=IO_IMPL,\
spark.sql.catalog.CATALOG_NAME.s3.client-factory-impl=org.apache.iceberg.aws.s3.DefaultS3FileIOAwsClientFactory,\
spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions"

Reemplaza IO_IMPL por la implementación de FileIO para el almacenamiento que contiene tus tablas:

• Tablas en Amazon S3: org.apache.iceberg.aws.s3.S3FileIO
• Tablas en Cloud Storage: org.apache.iceberg.gcp.gcs.GCSFileIO

La propiedad s3.client-factory-impl solo se aplica a las tablas de Amazon S3. Omite esta propiedad para las tablas de Cloud Storage.

Sin servidores

Crea una sesión interactiva sin servidores de Managed Service para Apache Spark en un notebook a través de la API de DataprocSparkSession con propiedades de configuración simplificadas (recomendado) o especificando manualmente las propiedades del catálogo.

Configuración simplificada con propiedades (recomendado)

En un notebook que tenga instalado el paquete dataproc-spark-connect, como un notebook de BigQuery Studio o de Vertex AI Workbench, pasa la propiedad del catálogo federado sin el prefijo dataproc: cuando crees la sesión:

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

catalog_name = "CATALOG_NAME"

session_config = Session()
session_config.runtime_config.version = 'RUNTIME_VERSION'

spark = DataprocSparkSession.builder.appName("APP_NAME") \
.dataprocSessionConfig(session_config) \
.config(f'dataproc.lakehouse.catalog.{catalog_name}', 'projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME') \
.config('spark.sql.defaultCatalog', catalog_name) \
.getOrCreate()

Reemplaza lo siguiente:

• CATALOG_NAME: Es un nombre para el catálogo local de Spark, por ejemplo, federated_catalog. Puede ser igual a la de FEDERATED_CATALOG_NAME.
• RUNTIME_VERSION: Es la versión del entorno de ejecución sin servidores de Managed Service para Apache Spark, por ejemplo, 3.0. Para conocer las versiones compatibles, consulta los requisitos que se mencionan más arriba en esta sección. Los roles de IAM necesarios dependen de la versión del entorno de ejecución.
• APP_NAME: Es el nombre de tu aplicación de Spark.
• PROJECT_ID: Es el ID del proyecto deGoogle Cloud .
• FEDERATED_CATALOG_NAME: Es el ID del catálogo federado que creaste en Antes de comenzar.

También puedes configurar un catálogo federado en un perfil de entorno de ejecución (plantilla de sesión) seleccionando Catálogo de tiempo de ejecución de Lakehouse en Configuración adicional > Configuración de Metastore en la página Crear un perfil de entorno de ejecución en la consola deGoogle Cloud , o estableciendo estas propiedades en una plantilla de sesión con gcloud CLI o la API de Dataproc.

Consulta el catálogo de Snowflake Horizon

Si consultas un catálogo de Snowflake Horizon, ten en cuenta lo siguiente:

• Almacenamiento de Snowflake: La configuración simplificada admite tablas de Snowflake almacenadas en Amazon S3. Para las tablas de Snowflake almacenadas en Cloud Storage, usa la configuración manual con io-impl establecido en org.apache.iceberg.gcp.gcs.GCSFileIO.

• Columnas STRING de Snowflake: Si encuentras errores cuando consultas columnas STRING de Snowflake en las que la política de serialización de almacenamiento de la tabla en Snowflake no es COMPATIBLE, puedes cambiar la política de serialización de almacenamiento a COMPATIBLE en Snowflake o inhabilitar la vectorización de Parquet en Spark agregando las siguientes llamadas .config() al compilador de sesiones:

  .config('spark.sql.iceberg.vectorization.enabled', 'false') \
  .config(f'spark.sql.catalog.{catalog_name}.table-override.read.parquet.vectorization.enabled', 'false')
      

Configuración manual

Si no usas la propiedad de configuración simplificada, especifica las propiedades del catálogo cuando crees la sesión. Para obtener una lista de los valores de los parámetros del proveedor de almacenamiento, consulta Parámetros de configuración de Spark.

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

catalog_name = "CATALOG_NAME"

session_config = Session()
session_config.runtime_config.version = 'RUNTIME_VERSION'

spark = DataprocSparkSession.builder.appName("APP_NAME") \
.dataprocSessionConfig(session_config) \
.config('spark.sql.defaultCatalog', catalog_name) \
.config(f'spark.sql.catalog.{catalog_name}', 'org.apache.iceberg.spark.SparkCatalog') \
.config(f'spark.sql.catalog.{catalog_name}.type', 'rest') \
.config(f'spark.sql.catalog.{catalog_name}.uri', 'https://biglake.googleapis.com/iceberg/v1/restcatalog') \
.config(f'spark.sql.catalog.{catalog_name}.warehouse', 'bl://projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME') \
.config(f'spark.sql.catalog.{catalog_name}.header.x-goog-user-project', 'PROJECT_ID') \
.config(f'spark.sql.catalog.{catalog_name}.rest.auth.type', 'org.apache.iceberg.gcp.auth.GoogleAuthManager') \
.config(f'spark.sql.catalog.{catalog_name}.header.X-Iceberg-Access-Delegation', 'vended-credentials') \
.config(f'spark.sql.catalog.{catalog_name}.io-impl', 'IO_IMPL') \
.config(f'spark.sql.catalog.{catalog_name}.s3.client-factory-impl', 'org.apache.iceberg.aws.s3.DefaultS3FileIOAwsClientFactory') \
.config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions') \
.getOrCreate()

Reemplaza RUNTIME_VERSION como se describe en el ejemplo de configuración simplificada que se encuentra antes en esta pestaña.

Reemplaza IO_IMPL por la implementación de FileIO para el almacenamiento que contiene tus tablas:

• Tablas en Amazon S3: org.apache.iceberg.aws.s3.S3FileIO
• Tablas en Cloud Storage: org.apache.iceberg.gcp.gcs.GCSFileIO

La propiedad s3.client-factory-impl solo se aplica a las tablas de Amazon S3. Omite esta línea para las tablas de Cloud Storage.

Consulta tablas federadas

Después de configurar tu sesión de Spark, puedes consultar tablas en tu catálogo federado con PySpark o Spark SQL.

Nota: La propiedad dataproc.lakehouse.catalog. registra el catálogo con Spark, pero no lo establece como el catálogo predeterminado. En los ejemplos anteriores, se establece la propiedad de clúster sin servidores spark.sql.defaultCatalog o spark:spark.sql.defaultCatalog, lo que te permite usar nombres de tablas no calificados. Si omites estas propiedades, debes calificar los nombres de las tablas con el nombre del catálogo, como se muestra en los siguientes ejemplos.

1. Lee una tabla federada con la API de DataFrame de PySpark:

   df = spark.table("CATALOG_NAME.NAMESPACE_NAME.TABLE_NAME")
   df.show(10, truncate=False)

   Reemplaza lo siguiente:

   • CATALOG_NAME: Es el nombre del catálogo local de Spark. Si configuras spark.sql.defaultCatalog como CATALOG_NAME, puedes omitir este prefijo.
   • NAMESPACE_NAME: Es el espacio de nombres (base de datos o esquema) en el catálogo federado.
   • TABLE_NAME: Es el nombre de la tabla en el catálogo federado.

2. Consulta la tabla federada con Spark SQL:

   spark.sql("""
       SELECT *
       FROM CATALOG_NAME.NAMESPACE_NAME.TABLE_NAME
       LIMIT 10
   """).show()

Une o materializa datos en catálogos federados y locales

Los catálogos federados son de solo lectura. Para unir datos remotos multinube con tablas locales o materializar los resultados de consultas federadas en Cloud Storage, conecta tu catálogo federado y un catálogo de lectura y escritura en el catálogo de tiempo de ejecución de Lakehouse con las propiedades dataproc.lakehouse.catalog. separadas por comas:

Clúster

Ejemplo de clúster de Managed Service para Apache Spark:

--properties="dataproc:dataproc.lakehouse.catalog.FEDERATED_SPARK_CATALOG=projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME,dataproc:dataproc.lakehouse.catalog.LOCAL_SPARK_CATALOG=projects/PROJECT_ID/catalogs/LOCAL_CATALOG_ID,spark:spark.sql.defaultCatalog=LOCAL_SPARK_CATALOG"

Sin servidores

Ejemplo de carga de trabajo por lotes sin servidores de Managed Service para Apache Spark:

--properties="dataproc.lakehouse.catalog.FEDERATED_SPARK_CATALOG=projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME,dataproc.lakehouse.catalog.LOCAL_SPARK_CATALOG=projects/PROJECT_ID/catalogs/LOCAL_CATALOG_ID,spark.sql.defaultCatalog=LOCAL_SPARK_CATALOG"

Sin servidores

Ejemplo de sesión interactiva sin servidores de Managed Service para Apache Spark:

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

session_config = Session()
session_config.runtime_config.version = 'RUNTIME_VERSION'

spark = DataprocSparkSession.builder.appName("APP_NAME") \
  .dataprocSessionConfig(session_config) \
  .config('dataproc.lakehouse.catalog.FEDERATED_SPARK_CATALOG', 'projects/PROJECT_ID/catalogs/FEDERATED_CATALOG_NAME') \
  .config('dataproc.lakehouse.catalog.LOCAL_SPARK_CATALOG', 'projects/PROJECT_ID/catalogs/LOCAL_CATALOG_ID') \
  .config('spark.sql.defaultCatalog', 'LOCAL_SPARK_CATALOG') \
  .getOrCreate()

Reemplaza RUNTIME_VERSION como se describe en el ejemplo de sesión interactiva que se encuentra más arriba en esta página.

Después de configurar estas propiedades, puedes consultar el catálogo federado y escribir los resultados en una tabla de tu catálogo de tiempo de ejecución de Lakehouse local.

La principal que ejecuta tu carga de trabajo es la cuenta de servicio para clústeres, entornos de ejecución sin servidores 2.3 y anteriores, y entornos de ejecución sin servidores 3.0 y posteriores con credenciales de cuenta de servicio. Para los entornos de ejecución sin servidores 3.0 y versiones posteriores con credenciales de usuario final (la opción predeterminada), la principal es tu cuenta de usuario.

Para garantizar que la principal que ejecuta tu carga de trabajo tenga los permisos necesarios para escribir datos de tablas en tu catálogo local, pídele a tu administrador que otorgue los siguientes roles de IAM a la principal que ejecuta tu carga de trabajo:

• Escribe datos de tabla en el modo de venta de credenciales: Editor de BigLake (roles/biglake.editor) en el proyecto que contiene el catálogo local
• Escribe datos de tablas en el modo de venta sin credenciales:
  • Editor de BigLake (roles/biglake.editor) en el proyecto que contiene el catálogo local
  • Usuario de objetos de almacenamiento (roles/storage.objectUser) en todos los buckets de Cloud Storage asociados con el catálogo local

En el siguiente ejemplo, se crea una tabla en tu catálogo local a partir de una tabla federada:

spark.sql("""
    CREATE TABLE IF NOT EXISTS LOCAL_SPARK_CATALOG.LOCAL_NAMESPACE.materialized_table
    USING iceberg
    AS SELECT *
    FROM FEDERATED_SPARK_CATALOG.REMOTE_NAMESPACE.REMOTE_TABLE
""")

¿Qué sigue?

• Crea un lakehouse con Spark y el catálogo de tiempo de ejecución de Lakehouse.
• Obtén más información para consultar datos remotos en el catálogo de tiempo de ejecución de Lakehouse.
• Lee Acerca del acceso a los datos multinube.
• Obtén más información sobre la venta de credenciales.