Utiliser BigLake Metastore avec Dataproc

Ce document explique comment utiliser le métastore BigLake 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 ou Apache Flink.

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 de BigLake Metastore et pourquoi vous devriez l'utiliser.

Rôles requis

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

• Créez un cluster Dataproc : Nœud de calcul Dataproc (roles/dataproc.worker) sur le compte de service par défaut Compute Engine du projet.
• Créez des tables BigLake Metastore dans Spark ou Flink :
  • Nœud de calcul Dataproc (roles/dataproc.worker) sur le compte de service de VM Dataproc du projet
  • Éditeur de données BigQuery (roles/bigquery.dataEditor) sur le compte de service de VM Dataproc du projet
  • Administrateur des objets de l'espace de stockage (roles/storage.objectAdmin) sur le compte de service de la VM Dataproc du projet
• Interroger les tables du metastore BigLake dans BigQuery :
  • Lecteur de données BigQuery (roles/bigquery.dataViewer) sur le projet
  • Utilisateur BigQuery (roles/bigquery.user) sur le projet
  • Lecteur d'objets Storage (roles/storage.objectViewer) sur le projet

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

Vous pouvez également obtenir les autorisations requises avec 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 BigLake, 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, tel que Spark ou Flink.
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 BigLake Metastore selon vos besoins, en fonction du moteur logiciel Open Source que vous utilisez.
5. Dans BigQuery, accédez à vos ressources BigLake Metastore et utilisez-les.

Connecter BigLake Metastore à Spark

Les instructions suivantes vous expliquent comment connecter Dataproc à BigLake Metastore à l'aide de Spark SQL interactif.

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

Pour connecter BigLake Metastore à Dataproc et Spark, vous devez utiliser le fichier JAR du plug-in de catalogue Iceberg pour BigLake Metastore.

Ce fichier est inclus par défaut dans la version d'image Dataproc 2.2. Si vos clusters Dataproc n'ont pas d'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 Iceberg pour BigLake Metastore.

Configurer un cluster Dataproc

Avant de vous connecter au métastore BigLake, 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 BigLake Metastore.

• L'accès privé à Google (PGA) doit être activé sur le sous-réseau de la région dans laquelle 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é.

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

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

Envoyer une tâche Spark

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

gcloud dataproc jobs submit spark-sql \
--project=PROJECT_ID \
--cluster=CLUSTER_NAME \
--region==REGION \
--jars=https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-spark-runtime-3.5_2.12/1.6.1/iceberg-spark-runtime-3.5_2.12-1.6.1.jar,gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.6.1-1.0.1-beta.jar \
--properties=spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog, \
spark.sql.catalog.CATALOG_NAME.catalog-impl=org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog, \
spark.sql.catalog.CATALOG_NAME.gcp_project=PROJECT_ID, \
spark.sql.catalog.CATALOG_NAME.gcp_location=LOCATION, \
spark.sql.catalog.CATALOG_NAME.warehouse=WAREHOUSE_DIRECTORY \
--execute="SPARK_SQL_COMMAND"

Connecter BigLake Metastore à Flink

Les instructions suivantes vous expliquent comment connecter Dataproc à BigLake Metastore à l'aide du client SQL Flink.

Installer le plug-in de catalogue et se connecter à une session Flink

Pour connecter BigLake Metastore à Flink, procédez comme suit:

1. Créez un cluster Dataproc avec le composant Flink facultatif activé et assurez-vous d'utiliser Dataproc 2.2 ou une version ultérieure.

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

   Accéder à la page Instances de VM

3. Dans la liste des instances de machine virtuelle, cliquez sur SSH pour vous connecter à une instance de VM Dataproc.

4. Configurez le plug-in de catalogue personnalisé Iceberg pour BigLake Metastore:

   FLINK_VERSION=1.17
   ICEBERG_VERSION=1.5.2
   
   cd /usr/lib/flink
   
   sudo wget -c https://repo.maven.apache.org/maven2/org/apache/iceberg/iceberg-flink-runtime-${FLINK_VERSION}/${ICEBERG_VERSION}/iceberg-flink-runtime-${FLINK_VERSION}-${ICEBERG_VERSION}.jar -P lib
   
   sudo gcloud storage cp gs://spark-lib/bigquery/iceberg-bigquery-catalog-${ICEBERG_VERSION}-1.0.1-beta.jar lib/

5. Démarrez la session Flink sur YARN:

   HADOOP_CLASSPATH=`hadoop classpath`
   
   sudo bin/yarn-session.sh -nm flink-dataproc -d
   
   sudo bin/sql-client.sh embedded \
     -s yarn-session

6. Créez un catalogue dans Flink:

   CREATE CATALOG CATALOG_NAME WITH (
     'type'='iceberg',
     'warehouse'='WAREHOUSE_DIRECTORY',
     'catalog-impl'='org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog',
     'gcp_project'='PROJECT_ID',
     'gcp_location'='LOCATION'
   );

   Remplacez les éléments suivants :

   • CATALOG_NAME: identifiant du catalogue Flink, associé à un catalogue BigLake Metastore.
   • WAREHOUSE_DIRECTORY: chemin d'accès de base au répertoire de l'entrepôt (dossier Cloud Storage dans lequel Flink crée des fichiers). Cette valeur commence par gs://.
   • PROJECT_ID: ID de projet du catalogue BigLake Metastore auquel le catalogue Flink est associé.
   • LOCATION: emplacement des ressources BigQuery.

Votre session Flink est désormais connectée au métastore BigLake, et vous pouvez exécuter des commandes SQL Flink.

Gérer les ressources de métastore BigLake

Maintenant que vous êtes connecté à BigLake Metastore, vous pouvez créer et afficher des ressources en fonction des métadonnées stockées dans BigLake Metastore.

Par exemple, essayez d'exécuter les commandes suivantes dans votre session SQL interactive Flink pour créer une base de données et une table Iceberg.

1. Utilisez le catalogue Iceberg personnalisé:

   USE CATALOG CATALOG_NAME;

   Remplacez CATALOG_NAME par l'identifiant de votre catalogue Flink.

2. Créez une base de données, qui crée un ensemble de données dans BigQuery:

   CREATE DATABASE IF NOT EXISTS DATABASE_NAME;

   Remplacez DATABASE_NAME par le nom de votre nouvelle base de données.

3. Utilisez la base de données que vous avez créée:

   USE DATABASE_NAME;

4. Créez une table Iceberg. L'exemple suivant crée un tableau des ventes:

   CREATE TABLE IF NOT EXISTS ICEBERG_TABLE_NAME (
       order_number BIGINT,
       price        DECIMAL(32,2),
       buyer        ROW<first_name STRING, last_name STRING>,
       order_time   TIMESTAMP(3)
   );

   Remplacez ICEBERG_TABLE_NAME par le nom de votre nouvelle table.

5. Afficher les métadonnées de table:

   DESCRIBE EXTENDED ICEBERG_TABLE_NAME;

6. Répertoriez les tables de la base de données:

   SHOW TABLES;

Ingérer des données dans votre table

Après avoir créé une table Iceberg dans la section précédente, vous pouvez utiliser Flink DataGen comme source de données pour insérer des données en temps réel dans votre table. Voici un exemple de ce workflow:

1. Créez une table temporaire à l'aide de DataGen:

   CREATE TEMPORARY TABLE DATABASE_NAME.TEMP_TABLE_NAME
   WITH (
       'connector' = 'datagen',
       'rows-per-second' = '10',
       'fields.order_number.kind' = 'sequence',
       'fields.order_number.start' = '1',
       'fields.order_number.end' = '1000000',
       'fields.price.min' = '0',
       'fields.price.max' = '10000',
       'fields.buyer.first_name.length' = '10',
       'fields.buyer.last_name.length' = '10'
   )
   LIKE DATABASE_NAME.ICEBERG_TABLE_NAME (EXCLUDING ALL);

   Remplacez les éléments suivants :

   • DATABASE_NAME: nom de la base de données dans laquelle stocker votre table temporaire.
   • TEMP_TABLE_NAME: nom de votre table temporaire.
   • ICEBERG_TABLE_NAME: nom de la table Iceberg que vous avez créée dans la section précédente.

2. Définissez le parallélisme sur 1:

   SET 'parallelism.default' = '1';

3. Définissez l'intervalle de contrôle:

   SET 'execution.checkpointing.interval' = '10second';

4. Définissez le point de contrôle:

   SET 'state.checkpoints.dir' = 'hdfs:///flink/checkpoints';

5. Démarrez la tâche de streaming en temps réel:

   INSERT INTO ICEBERG_TABLE_NAME SELECT * FROM TEMP_TABLE_NAME;

   Le résultat ressemble à ce qui suit :

   [INFO] Submitting SQL update statement to the cluster...
   [INFO] SQL update statement has been successfully submitted to the cluster:
   Job ID: 0de23327237ad8a811d37748acd9c10b
   

6. Pour vérifier l'état de la tâche de streaming, procédez comme suit:

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

      accéder aux clusters

   2. Sélectionnez votre cluster.

   3. Cliquez sur l'onglet Interfaces Web.

   4. Cliquez sur le lien YARN ResourceManager (Gestionnaire de ressources YARN).

   5. Dans l'interface YARN ResourceManager, recherchez votre session Flink, puis cliquez sur le lien ApplicationMaster sous Tracking UI (UI de suivi).

   6. Dans la colonne État, vérifiez que l'état de votre tâche est En cours d'exécution.

7. Interroger des données de streaming dans le client SQL Flink:

   SELECT * FROM ICEBERG_TABLE_NAME
   /*+ OPTIONS('streaming'='true', 'monitor-interval'='3s')*/
   ORDER BY order_time desc
   LIMIT 20;

8. Interroger des données en flux continu dans BigQuery:

   SELECT * FROM `DATABASE_NAME.ICEBERG_TABLE_NAME`
   ORDER BY order_time desc
   LIMIT 20;

9. Arrêtez la tâche de streaming dans le client SQL Flink:

   STOP JOB 'JOB_ID';

   Remplacez JOB_ID par l'ID de tâche qui s'affiche dans la sortie lorsque vous avez créé la tâche de streaming.

Étape suivante

• Configurez les fonctionnalités facultatives de BigLake Metastore.
• Afficher et interroger des tables à partir de Spark dans BigQuery