Utiliser le métastore BigQuery avec Dataproc

.

Ce document explique comment utiliser le métastore BigQuery avec Dataproc sur Compute Engine. Cette connexion vous fournit un métastore unique et partagé qui fonctionne avec les moteurs logiciels Open Source, tels qu'Apache Spark.

Avant de commencer

  1. Activez la facturation pour votre projet Google Cloud . Découvrez comment vérifier si la facturation est activée sur un projet.

  2. Activez les API BigQuery et Dataproc.

    Activer les API

  3. Facultatif: Découvrez le fonctionnement du métastore BigQuery et pourquoi vous devriez l'utiliser.

Rôles requis

Pour obtenir les autorisations nécessaires pour utiliser Spark et Dataproc avec le métastore BigQuery en tant que magasin de métadonnées, demandez à votre administrateur de vous accorder les rôles IAM suivants:

Pour en savoir plus sur l'attribution de rôles, consultez la page Gérer l'accès aux projets, aux dossiers et aux organisations.

Vous pouvez également obtenir les autorisations requises via des rôles personnalisés ou d'autres rôles prédéfinis.

Workflow général

Pour utiliser Dataproc sur Compute Engine avec le métastore BigQuery, procédez comme suit:

  1. Créez un cluster Dataproc ou configurez un cluster existant.
  2. Connectez-vous au moteur logiciel Open Source de votre choix, comme Spark.
  3. Utilisez un fichier JAR pour installer le plug-in de catalogue Apache Iceberg sur le cluster.
  4. Créez et gérez vos ressources de métastore BigQuery selon vos besoins, en fonction du moteur logiciel Open Source que vous utilisez.
  5. Dans BigQuery, accédez à vos ressources de métastore BigQuery et utilisez-les.

Connecter le métastore BigQuery à Spark

Les instructions suivantes vous expliquent comment connecter Dataproc au métastore BigQuery à l'aide de Spark SQL interactif.

Télécharger le plug-in de catalogue Iceberg

Pour connecter le métastore BigQuery à Dataproc et Spark, vous devez utiliser le fichier JAR du plug-in de catalogue Iceberg du métastore BigQuery.

Ce fichier est inclus par défaut dans la version 2.2 de l'image Dataproc. Si vos clusters Dataproc n'ont pas accès direct à Internet, vous devez télécharger le plug-in et l'importer dans un bucket Cloud Storage auquel votre cluster Dataproc peut accéder.

Téléchargez le plug-in de catalogue Apache Iceberg pour le métastore BigQuery.

Configurer un cluster Dataproc

Avant de vous connecter au métastore BigQuery, vous devez configurer un cluster Dataproc.

Pour ce faire, vous pouvez créer un cluster ou utiliser un cluster existant. Vous utiliserez ensuite ce cluster pour exécuter Spark SQL interactif et gérer vos ressources Metastore BigQuery.

  • L'accès privé à Google (PGA) doit être activé pour le sous-réseau de la région où le cluster est créé. Par défaut, les VM de cluster Dataproc créées avec une version d'image 2.2 (par défaut) ou ultérieure ne comportent que des adresses IP internes. Pour permettre aux VM du cluster de communiquer avec les API Google, activez l'accès privé à Google sur le sous-réseau réseau default (ou le nom de réseau spécifié par l'utilisateur, le cas échéant) dans la région où le cluster est créé.

  • Si vous souhaitez exécuter l'exemple d'interface Web Zeppelin de ce guide, vous devez utiliser ou créer un cluster Dataproc avec le composant facultatif Zeppelin activé.

Nouveau cluster

Pour créer un cluster Dataproc, exécutez la commande gcloud dataproc clusters create suivante. Cette configuration contient les paramètres dont vous avez besoin pour utiliser le métastore BigQuery.

gcloud dataproc clusters create CLUSTER_NAME \
  --project=PROJECT_ID \
  --region=LOCATION \
  --optional-components=ZEPPELIN \
  --enable-component-gateway \
  --single-node

Remplacez les éléments suivants :

  • CLUSTER_NAME: nom de votre cluster Dataproc.
  • PROJECT_ID: ID du projet Google Cloud dans lequel vous créez le cluster.
  • LOCATION: région Google Cloud dans laquelle vous créez le cluster.

Cluster existant

Pour configurer un cluster existant, ajoutez l'environnement d'exécution Iceberg Spark suivant à votre cluster.

org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.5.2

Vous pouvez ajouter l'environnement d'exécution à l'aide de l'une des options suivantes:

Envoyer une tâche Spark

Pour envoyer une tâche Spark, utilisez l'une des méthodes suivantes:

CLI gcloud

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.5.2/iceberg-spark-runtime-3.5_2.12-1.5.2.jar,gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.5.2-1.0.0-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"

Remplacez les éléments suivants :

  • PROJECT_ID: ID du projet Google Cloud qui contient le cluster Dataproc.
  • CLUSTER_NAME: nom du cluster Dataproc que vous utilisez pour exécuter la tâche SQL Spark.
  • REGION: région Compute Engine où se trouve votre cluster.
  • LOCATION: emplacement des ressources BigQuery.
  • CATALOG_NAME: nom du catalogue Spark que vous utilisez avec votre tâche SQL.
  • WAREHOUSE_DIRECTORY: le dossier Cloud Storage contenant votre entrepôt de données. Cette valeur commence par gs://.
  • SPARK_SQL_COMMAND: la requête Spark SQL que vous souhaitez exécuter. Cette requête inclut les commandes permettant de créer vos ressources. Par exemple, pour créer un espace de noms et une table.

Interactive Spark

Se connecter à Spark et installer le plug-in de catalogue

Pour installer le plug-in de catalogue pour le métastore BigQuery, connectez-vous à votre cluster Dataproc à l'aide de SSH.

  1. Dans la console Google Cloud , accédez à la page Instances de VM.

  2. Pour vous connecter à une instance de VM Dataproc, cliquez sur SSH dans la liste des instances de machines virtuelles. Le résultat ressemble à ce qui suit :

    Connected, host fingerprint: ssh-rsa ...
Linux cluster-1-m 3.16.0-0.bpo.4-amd64 ...
...
example-cluster@cluster-1-m:~$

  3. Dans le terminal, exécutez la commande d'initialisation du métastore BigQuery suivante:

    spark-sql \
--jars https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-spark-runtime-3.5_2.12/1.5.2/iceberg-spark-runtime-3.5_2.12-1.5.2.jar,gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.5.2-1.0.0-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

    Remplacez les éléments suivants :

    • CATALOG_NAME: nom du catalogue Spark que vous utilisez avec votre tâche SQL.
    • PROJECT_ID: ID de projet Google Cloud du catalogue de métadonnées BigQuery auquel votre catalogue Spark est associé.
    • LOCATION: emplacement du métastore BigQuery dans Google Cloud .
    • WAREHOUSE_DIRECTORY: le dossier Cloud Storage contenant votre entrepôt de données. Cette valeur commence par gs://.

    Une fois que vous vous êtes connecté à un cluster, le terminal Spark affiche l'invite spark-sql.

    spark-sql (default)>

Gérer les ressources de métastore BigQuery

Vous êtes maintenant connecté au métastore BigQuery. Vous pouvez afficher vos ressources existantes ou en créer en fonction de vos métadonnées stockées dans le métastore BigQuery.

Par exemple, essayez d'exécuter les commandes suivantes dans la session interactive Spark SQL pour créer un espace de noms et une table Iceberg.

  • Utilisez le catalogue Iceberg personnalisé:

    USE `CATALOG_NAME`;

  • Créez un espace de noms :

    CREATE NAMESPACE IF NOT EXISTS NAMESPACE_NAME;

  • Utilisez l'espace de noms créé:

    USE NAMESPACE_NAME;

  • Créez un tableau iceberg:

    CREATE TABLE TABLE_NAME (id int, data string) USING ICEBERG;

  • Insérer une ligne de tableau:

    INSERT INTO TABLE_NAME VALUES (1, "first row");

  • Ajoutez une colonne de tableau:

    ALTER TABLE TABLE_NAME ADD COLUMNS (newDoubleCol double);

  • Afficher les métadonnées d'une table:

    DESCRIBE EXTENDED TABLE_NAME;

  • Répertoriez les tables de l'espace de noms:

    SHOW TABLES;

Notebook Zeppelin

  1. Dans la console Google Cloud , accédez à la page Clusters Dataproc.

    Accéder aux clusters Dataproc

  2. Cliquez sur le nom du cluster que vous souhaitez utiliser.

    La page Détails du cluster s'ouvre.

  3. Dans le menu de navigation, cliquez sur Interfaces Web.

  4. Sous Balise de composant, cliquez sur Zeppelin. La page du notebook Zeppelin s'ouvre.

  5. Dans le menu de navigation, cliquez sur Carnet, puis sur + Créer une note.

  6. Dans la boîte de dialogue, saisissez un nom de notebook. Laissez Spark sélectionné comme interpréteur par défaut.

  7. Cliquez sur Créer. Un nouveau notebook est créé.

  8. Dans le notebook, cliquez sur le menu des paramètres, puis sur Interpréteur.

  9. Dans le champ Interpréteurs de recherche, recherchez Spark.

  10. Cliquez sur Modifier.

  11. Dans le champ Spark.jars, saisissez l'URI du fichier JAR Spark.

    https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-spark-runtime-3.5_2.12/1.5.2/iceberg-spark-runtime-3.5_2.12-1.5.2.jar,gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.5.2-1.0.0-beta.jar

  12. Cliquez sur Enregistrer.

  13. Cliquez sur OK.

  14. Copiez le code PySpark suivant dans votre notebook Zeppelin.

    %pyspark
from pyspark.sql import SparkSession
spark = SparkSession.builder \
.appName("BigQuery 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("select version()").show()
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;")
spark.sql("DESCRIBE TABLE_NAME;").show()

    Remplacez les éléments suivants :

    • CATALOG_NAME: nom du catalogue Spark à utiliser pour la tâche SQL.
    • PROJECT_ID: ID du projet Google Cloud contenant le cluster Dataproc.
    • WAREHOUSE_DIRECTORY: le dossier Cloud Storage contenant votre entrepôt de données. Cette valeur commence par gs://.
    • NAMESPACE_NAME: nom de l'espace de noms qui fait référence à votre table Spark.
    • WAREHOUSE_DIRECTORY: URI du dossier Cloud Storage dans lequel votre entrepôt de données est stocké.
    • TABLE_NAME: nom de la table Spark.

  15. Cliquez sur l'icône d'exécution ou appuyez sur Shift-Enter pour exécuter le code. Une fois la tâche terminée, le message d'état indique "Spark Job Finished" (Tâche Spark terminée), et le résultat affiche le contenu de la table:

Étape suivante