Interroger des catalogues fédérés avec Spark et le catalogue d'environnements d'exécution Lakehouse

Aperçu

Cette fonctionnalité est soumise aux "Conditions des offres de pré-DG" de la section "Conditions générales du service" des Conditions spécifiques du service. Les fonctionnalités pré-DG sont disponibles "en l'état" et peuvent avoir une prise en charge limitée. Pour en savoir plus, consultez les descriptions des étapes de lancement.

Après avoir configuré l'accès aux données inter-cloud dans le catalogue d'environnements d'exécution Lakehouse, vous pouvez interroger des tables Iceberg distantes à partir de Managed Service pour Apache Spark sans migrer de fichiers ni gérer d'identifiants de stockage cloud distincts.

Le catalogue d'environnements d'exécution Lakehouse gère la synchronisation des métadonnées et l'accès sécurisé aux données via la distribution d'identifiants. Lorsque Spark interroge une table fédérée, le catalogue d'environnements d'exécution Lakehouse traduit les métadonnées du catalogue et émet des identifiants limités dans le temps et dans le champ d'application pour que Spark puisse lire les données d'Amazon S3 ou de Cloud Storage, en traitant les tables comme si elles étaient locales dans votre environnementGoogle Cloud .

Types de catalogues fédérés compatibles

Managed Service pour Apache Spark permet l'interrogation des types de catalogues fédérés suivants via le point de terminaison du catalogue REST Iceberg du catalogue d'environnements d'exécution Lakehouse :

Type de catalogue fédéré	Backends de stockage compatibles	Guide de configuration dans le catalogue d'environnements d'exécution Lakehouse
AWS Glue Data Catalog	Amazon S3 (s3:// : catalogue de compte AWS par défaut ou bucket de tables Amazon S3)	Se connecter à AWS Glue
Databricks Unity Catalog	Emplacements externes Amazon S3 (s3://) ou Cloud Storage (gs://)	Se connecter à Databricks Unity Catalog
Catalogue Snowflake Horizon	Amazon S3 (s3://) ou Cloud Storage (gs://)	Se connecter au catalogue Snowflake Horizon

Pour en savoir plus sur les exigences de préparation du catalogue et les définitions des paramètres Spark sous-jacents pour chaque fournisseur de stockage, consultez Interroger des données distantes dans le catalogue d'exécution Lakehouse et Paramètres de configuration Spark.

Avant de commencer

Connectez-vous à votre compte Google Cloud . Si vous débutez sur Google Cloud, créez un compte pour évaluer les performances de nos produits en conditions réelles. Les nouveaux clients bénéficient également de 300 $ de crédits sans frais pour exécuter, tester et déployer des charges de travail.

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. Assurez-vous de disposer de tables Iceberg dans votre catalogue distant.
2. Configurez une connexion cross-cloud et un catalogue fédéré dans le catalogue d'environnements d'exécution Lakehouse pour votre fournisseur de catalogue distant :
   • AWS Glue Data Catalog
   • Databricks Unity Catalog
   • Catalogue Snowflake Horizon

Rôles IAM (Identity and Access Management) requis

Certains rôles IAM sont requis pour exécuter les exemples de cette page. En fonction des règles d'administration, ces rôles peuvent déjà avoir été accordés. Pour vérifier les attributions de rôles, consultez Devez-vous attribuer des rôles ?. Pour en savoir plus sur l'attribution de rôles, consultez Gérer l'accès aux projets, aux dossiers et aux organisations.

Les rôles dont vous avez besoin dépendent de la façon dont votre charge de travail s'exécute. Les clusters Managed Service pour Apache Spark et les runtimes sans serveur Managed Service pour Apache Spark 2.3 et versions antérieures exécutent les charges de travail en tant que compte de service. Les runtimes sans serveur Managed Service for Apache Spark 3.0 et versions ultérieures exécutent les charges de travail avec vos identifiants utilisateur finaux par défaut. Vous devez donc accorder des rôles d'accès aux données à votre compte utilisateur au lieu d'un compte de service. Si vous spécifiez un compte de service personnalisé pour un environnement d'exécution 3.0 ou version ultérieure, la charge de travail s'exécute avec les identifiants du compte de service. Pour en savoir plus, consultez Rôles des déploiements sans serveur.

Rôles utilisateur

Pour obtenir les autorisations nécessaires pour créer un cluster Managed Service pour Apache Spark, envoyer une charge de travail par lot ou créer une session interactive, demandez à votre administrateur de vous accorder les rôles IAM suivants :

• Utilisez un cluster ou un environnement d'exécution sans serveur 2.3 ou version antérieure :
  • Éditeur Dataproc (roles/dataproc.editor) sur le projet
  • Utilisateur du compte de service (roles/iam.serviceAccountUser) sur le compte de service Compute Engine par défaut ou un compte de service personnalisé
• Utilisez un environnement d'exécution sans serveur 3.0 ou version ultérieure : Éditeur Dataproc sans serveur (roles/dataproc.serverlessEditor) sur le projet
• Lire les tables fédérées avec les identifiants de l'utilisateur final sur un environnement d'exécution sans serveur 3.0 ou version ultérieure : Lecteur BigLake (roles/biglake.viewer) sur le projet contenant le catalogue fédéré
• Utilisez les identifiants du compte de service sur un environnement d'exécution sans serveur 3.0 ou version ultérieure : Utilisateur du compte de service (roles/iam.serviceAccountUser) sur le compte de service personnalisé

Rôles de compte de service

Attribuez les rôles suivants au compte de service qui exécute votre charge de travail. Les runtimes sans serveur Managed Service pour Apache Spark 3.0 et versions ultérieures n'ont besoin de ces rôles que si la charge de travail s'exécute avec des identifiants de compte de service.

Pour vous assurer que le compte de service Compute Engine par défaut ou un compte de service personnalisé dispose des autorisations nécessaires pour exécuter des jobs Managed Service pour Apache Spark et lire des tables fédérées, demandez à votre administrateur d'accorder les rôles IAM suivants au compte de service Compute Engine par défaut ou à un compte de service personnalisé :

Important : Vous devez attribuer ces rôles au compte de service Compute Engine par défaut ou à un compte de service personnalisé, et non à votre compte utilisateur. Si vous ne les accordez pas au bon compte principal, vous risquez de rencontrer des erreurs d'autorisation.

• Exécuter des jobs Managed Service pour Apache Spark sur un cluster ou un environnement d'exécution sans serveur 2.3 ou version antérieure : Nœud de calcul Dataproc (roles/dataproc.worker) sur le projet
• Exécutez des jobs Managed Service pour Apache Spark sur un environnement d'exécution sans serveur 3.0 ou version ultérieure :
  • Nœud Dataproc sans serveur (roles/dataproc.serverlessNode) sur le projet
  • Agent de service de nœud Resource Manager pour Dataproc (roles/dataprocrm.nodeServiceAgent) sur le projet
• Découvrez et lisez les métadonnées des tables fédérées au moment de l'exécution : Lecteur BigLake (roles/biglake.viewer) sur le projet contenant le catalogue fédéré

Pour vous assurer que l'agent de service Managed Service pour Apache Spark ([e-mail]) dispose des autorisations nécessaires pour résoudre la propriété de configuration simplifiée dataproc.lakehouse.catalog. lorsque vous créez un cluster, envoyez une charge de travail par lot ou créez une session interactive, demandez à votre administrateur d'accorder le rôle IAM Lecteur BigLake (roles/biglake.viewer) à l'agent de service Managed Service pour Apache Spark ([e-mail]) sur le projet contenant le catalogue fédéré.

Important : Vous devez attribuer ce rôle à l'agent de service Managed Service pour Apache Spark ([e-mail]), et non à votre compte utilisateur. Si vous ne l'accordez pas au bon compte principal, vous risquez de rencontrer des erreurs d'autorisation.

Configurer la connectivité réseau sortante pour Amazon S3

Lorsque vous interrogez des catalogues fédérés basés sur Amazon S3, les charges de travail par lot sans serveur et les sessions interactives de Managed Service pour Apache Spark, ainsi que les clusters Managed Service pour Apache Spark avec adresse IP interne, ne peuvent pas se connecter aux points de terminaison Amazon S3 avec leur configuration réseau par défaut. Provisionnez un routeur Cloud Router et Cloud NAT sur votre réseau VPC de charge de travail pour autoriser le trafic sortant vers 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

Remplacez les éléments suivants :

• NETWORK_NAME : réseau VPC de votre cluster, charge de travail par lot ou session interactive Managed Service pour Apache Spark, par exemple default.
• REGION : région Google Cloud pour votre cluster Managed Service pour Apache Spark, votre charge de travail par lot ou votre session interactive.

Configurer Managed Service pour Apache Spark et interroger des tables fédérées

Vous pouvez configurer un cluster Managed Service pour Apache Spark, une charge de travail par lot sans serveur ou une session interactive pour vous connecter à votre catalogue fédéré dans le catalogue d'environnements d'exécution Lakehouse. Ces ressources Managed Service pour Apache Spark sont compatibles avec la propriété de configuration simplifiée dataproc.lakehouse.catalog.CATALOG_NAME (recommandée), qui lit la définition du catalogue fédéré à partir du catalogue d'environnements d'exécution Lakehouse et génère les propriétés Spark du catalogue REST Iceberg requises, y compris les en-têtes de stockage io-impl et de distribution d'identifiants (la forme de cluster de la propriété ajoute le préfixe dataproc:).

La configuration simplifiée et la configuration manuelle utilisent l'autorisation Google pour Iceberg (GoogleAuthManager), qui nécessite l'une des versions d'image de cluster ou d'environnement d'exécution sans serveur suivantes :

• Clusters Managed Service pour Apache Spark : versions d'image 2.2 (2.2.65 et ultérieures) ou versions d'image 2.3 (2.3.11 et ultérieures).
• Charges de travail par lot et sessions interactives sans serveur Managed Service pour Apache Spark : environnements d'exécution 2.2.60 et ultérieurs, environnements d'exécution 2.3.10 et ultérieurs, ou environnements d'exécution 3.0.

Pour les charges de travail par lot et les sessions interactives sans serveur, les rôles requis dépendent de la version du runtime. Les environnements d'exécution 3.0 et versions ultérieures utilisent par défaut les identifiants de l'utilisateur final. Pour en savoir plus, consultez Rôles IAM requis et Rôles pour les déploiements sans serveur.

Cluster

Créez un cluster Managed Service pour Apache Spark avec le composant facultatif Iceberg à l'aide de propriétés de configuration simplifiées (recommandé) ou en spécifiant manuellement les propriétés du catalogue.

Configuration simplifiée à l'aide de propriétés (recommandé)

Créez le cluster avec la propriété de catalogue fédéré :

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"

Remplacez les éléments suivants :

• CLUSTER_NAME : nom de votre cluster.
• PROJECT_ID : ID de votre projetGoogle Cloud .
• REGION : région Google Clouddu cluster, par exemple us-central1.
• IMAGE_VERSION : version de l'image Managed Service pour Apache Spark (par exemple, 2.3-debian12). Pour connaître les versions compatibles, consultez les conditions requises plus haut dans cette section.
• CATALOG_NAME : nom du catalogue Spark local, par exemple federated_catalog. Il peut être identique à FEDERATED_CATALOG_NAME.
• FEDERATED_CATALOG_NAME : ID du catalogue fédéré que vous avez créé dans la section Avant de commencer.

Remarque : Vous pouvez transmettre plusieurs catalogues séparés par une virgule à l'option --properties. Chaque propriété de catalogue de cluster nécessite le préfixe dataproc:.

Interroger le catalogue Snowflake Horizon

Si vous interrogez un catalogue Snowflake Horizon, notez les points suivants :

• Stockage Snowflake : la configuration simplifiée est compatible avec les tables Snowflake stockées dans Amazon S3. Pour les tables Snowflake stockées dans Cloud Storage, utilisez la configuration manuelle avec io-impl défini sur org.apache.iceberg.gcp.gcs.GCSFileIO.

• Colonnes Snowflake STRING : si vous rencontrez des erreurs lors de l'interrogation des colonnes Snowflake STRING où la règle de sérialisation du stockage de la table dans Snowflake n'est pas COMPATIBLE, vous pouvez modifier la règle de sérialisation du stockage en COMPATIBLE dans Snowflake ou désactiver la vectorisation Parquet dans Spark en ajoutant les propriétés suivantes à l'option --properties :

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

Dans votre notebook Jupyter ou votre application PySpark, créez le SparkSession sans spécifier de configuration de catalogue :

from pyspark.sql import SparkSession

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

Remplacez APP_NAME par le nom de votre application Spark.

Configuration manuelle

Si vous n'utilisez pas la propriété de configuration simplifiée, créez le cluster comme décrit précédemment dans cette section, sans transmettre la propriété dataproc:dataproc.lakehouse.catalog. à l'indicateur --properties, puis configurez manuellement votre SparkSession. Pour obtenir la liste des valeurs des paramètres du fournisseur de stockage, consultez Paramètres de configuration 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()

Remplacez IO_IMPL par l'implémentation FileIO de l'espace de stockage contenant vos tables :

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

La propriété s3.client-factory-impl ne s'applique qu'aux tables d'Amazon S3. Omettez cette ligne pour les tables Cloud Storage.

Sans serveur

Envoyez une charge de travail par lot PySpark à Managed Service pour Apache Spark sans serveur à l'aide de propriétés de configuration simplifiées (recommandé) ou en spécifiant manuellement les propriétés du catalogue.

Configuration simplifiée à l'aide de propriétés (recommandé)

Transmettez la propriété de catalogue fédéré sans le préfixe dataproc: lorsque vous envoyez la charge de travail par lot :

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"

Remplacez les éléments suivants :

• PYSPARK_FILE : chemin d'accès Cloud Storage gs:// vers le fichier de votre application PySpark.
• PROJECT_ID : ID de votre projetGoogle Cloud .
• REGION : région de la charge de travail par lot.
• RUNTIME_VERSION : version d'exécution sans serveur de Managed Service pour Apache Spark (par exemple, 3.0). Pour connaître les versions compatibles, consultez les exigences plus haut dans cette section. Les rôles IAM requis dépendent de la version du runtime.
• CATALOG_NAME : nom du catalogue Spark local, par exemple federated_catalog. Il peut être identique à FEDERATED_CATALOG_NAME.
• FEDERATED_CATALOG_NAME : ID du catalogue fédéré que vous avez créé dans la section Avant de commencer.

Remarque : Vous pouvez transmettre plusieurs catalogues séparés par une virgule à l'indicateur --properties.

Interroger le catalogue Snowflake Horizon

Si vous interrogez un catalogue Snowflake Horizon, notez les points suivants :

• Stockage Snowflake : la configuration simplifiée est compatible avec les tables Snowflake stockées dans Amazon S3. Pour les tables Snowflake stockées dans Cloud Storage, utilisez la configuration manuelle avec io-impl défini sur org.apache.iceberg.gcp.gcs.GCSFileIO.

• Colonnes Snowflake STRING : si vous rencontrez des erreurs lors de l'interrogation des colonnes Snowflake STRING où la règle de sérialisation du stockage de la table dans Snowflake n'est pas COMPATIBLE, vous pouvez modifier la règle de sérialisation du stockage sur COMPATIBLE dans Snowflake ou désactiver la vectorisation Parquet dans Spark en ajoutant les propriétés suivantes à l'option --properties :

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

Dans le fichier d'application PySpark, créez SparkSession sans spécifier de configurations de catalogue :

from pyspark.sql import SparkSession

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

Configuration manuelle

Si vous n'utilisez pas la propriété de configuration simplifiée, spécifiez les propriétés du catalogue avec l'option --properties de la commande gcloud dataproc batches submit. Pour obtenir la liste des valeurs des paramètres du fournisseur de stockage, consultez Paramètres de configuration 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"

Remplacez IO_IMPL par l'implémentation FileIO pour le stockage contenant vos tables :

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

La propriété s3.client-factory-impl ne s'applique qu'aux tables d'Amazon S3. Omettez cette propriété pour les tables Cloud Storage.

Sans serveur

Créez une session interactive sans serveur Managed Service pour Apache Spark dans un notebook via l'API DataprocSparkSession à l'aide de propriétés de configuration simplifiées (recommandé) ou en spécifiant manuellement les propriétés du catalogue.

Configuration simplifiée à l'aide de propriétés (recommandé)

Dans un notebook sur lequel le package dataproc-spark-connect est installé, tel qu'un notebook BigQuery Studio ou un notebook Vertex AI Workbench, transmettez la propriété de catalogue fédéré sans le préfixe dataproc: lorsque vous créez la session :

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()

Remplacez les éléments suivants :

• CATALOG_NAME : nom du catalogue Spark local, par exemple federated_catalog. Il peut être identique à FEDERATED_CATALOG_NAME.
• RUNTIME_VERSION : version d'exécution sans serveur de Managed Service pour Apache Spark (par exemple, 3.0). Pour connaître les versions compatibles, consultez les exigences plus haut dans cette section. Les rôles IAM requis dépendent de la version du runtime.
• APP_NAME : nom de votre application Spark.
• PROJECT_ID : ID de votre projetGoogle Cloud .
• FEDERATED_CATALOG_NAME : ID du catalogue fédéré que vous avez créé dans la section Avant de commencer.

Vous pouvez également définir ces propriétés dans un modèle de session.

Interroger le catalogue Snowflake Horizon

Si vous interrogez un catalogue Snowflake Horizon, notez les points suivants :

• Stockage Snowflake : la configuration simplifiée est compatible avec les tables Snowflake stockées dans Amazon S3. Pour les tables Snowflake stockées dans Cloud Storage, utilisez la configuration manuelle avec io-impl défini sur org.apache.iceberg.gcp.gcs.GCSFileIO.

• Colonnes Snowflake STRING : si vous rencontrez des erreurs lors de l'interrogation des colonnes Snowflake STRING où la règle de sérialisation du stockage de la table dans Snowflake n'est pas COMPATIBLE, vous pouvez remplacer la règle de sérialisation du stockage par COMPATIBLE dans Snowflake ou désactiver la vectorisation Parquet dans Spark en ajoutant les appels .config() suivants au compilateur de session :

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

Configuration manuelle

Si vous n'utilisez pas la propriété de configuration simplifiée, spécifiez les propriétés du catalogue lorsque vous créez la session. Pour obtenir la liste des valeurs des paramètres du fournisseur de stockage, consultez Paramètres de configuration 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()

Remplacez RUNTIME_VERSION comme décrit dans l'exemple de configuration simplifiée plus haut dans cet onglet.

Remplacez IO_IMPL par l'implémentation FileIO pour le stockage contenant vos tables :

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

La propriété s3.client-factory-impl ne s'applique qu'aux tables Amazon S3. Omettez cette ligne pour les tables Cloud Storage.

Interroger des tables fédérées

Après avoir configuré votre session Spark, vous pouvez interroger les tables de votre catalogue fédéré à l'aide de PySpark ou de Spark SQL.

Remarque : La propriété dataproc.lakehouse.catalog. enregistre le catalogue auprès de Spark, mais ne le définit pas comme catalogue par défaut. Les exemples précédents définissent la propriété de cluster spark.sql.defaultCatalog sans serveur ou spark:spark.sql.defaultCatalog, ce qui vous permet d'utiliser des noms de table non qualifiés. Si vous omettez ces propriétés, vous devez qualifier les noms de tables avec le nom du catalogue, comme illustré dans les exemples suivants.

1. Lisez une table fédérée à l'aide de l'API PySpark DataFrame :

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

   Remplacez les éléments suivants :

   • CATALOG_NAME : nom du catalogue Spark local. Si vous définissez spark.sql.defaultCatalog sur CATALOG_NAME, vous pouvez omettre ce préfixe.
   • NAMESPACE_NAME : espace de noms (base de données ou schéma) dans le catalogue fédéré.
   • TABLE_NAME : nom de la table dans le catalogue fédéré.

2. Interrogez la table fédérée à l'aide de Spark SQL :

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

Joindre ou matérialiser des données dans des catalogues fédérés et locaux

Les catalogues fédérés sont en lecture seule. Pour joindre des données cross-cloud distantes à des tables locales ou matérialiser des résultats de requêtes fédérées dans Cloud Storage, associez votre catalogue fédéré et un catalogue en lecture et en écriture dans le catalogue d'environnements d'exécution Lakehouse à l'aide des propriétés dataproc.lakehouse.catalog. séparées par une virgule :

Cluster

Exemple de cluster Managed Service pour 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"

Sans serveur

Exemple de charge de travail par lot sans serveur Managed Service pour 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"

Sans serveur

Exemple de session interactive sans serveur Managed Service pour 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()

Remplacez RUNTIME_VERSION comme décrit dans l'exemple de session interactive plus haut sur cette page.

Après avoir défini ces propriétés, vous pouvez interroger le catalogue fédéré et écrire les résultats dans une table de votre catalogue d'exécution Lakehouse local.

Le compte principal qui exécute votre charge de travail est le compte de service pour les clusters, les environnements d'exécution sans serveur 2.3 et versions antérieures, et les environnements d'exécution sans serveur 3.0 et versions ultérieures avec des identifiants du compte de service. Pour les environnements d'exécution sans serveur 3.0 et versions ultérieures avec des identifiants d'utilisateur final (par défaut), le compte principal est votre compte utilisateur.

Pour vous assurer que le compte principal qui exécute votre charge de travail dispose des autorisations nécessaires pour écrire des données de table dans votre catalogue local, demandez à votre administrateur d'accorder les rôles IAM suivants au compte principal qui exécute votre charge de travail :

• Écrire des données de table en mode de distribution des identifiants : Éditeur BigLake (roles/biglake.editor) sur le projet contenant le catalogue local
• Écrivez les données de la table en mode de distribution sans identifiants :
  • Éditeur BigLake (roles/biglake.editor) sur le projet contenant le catalogue local
  • Utilisateur d'objets Storage (roles/storage.objectUser) sur tous les buckets Cloud Storage associés au catalogue local

L'exemple suivant crée une table dans votre catalogue local à partir d'une table fédérée :

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
""")

Étapes suivantes

• Créez un lakehouse avec Spark et le catalogue d'environnements d'exécution Lakehouse.
• En savoir plus sur l'interrogation de données à distance dans le catalogue d'environnements d'exécution Lakehouse
• Consultez À propos de l'accès aux données multicloud.
• En savoir plus sur la distribution d'identifiants