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 à l'utilisation de Spark ou Flink et de Dataproc avec le metastore BigLake comme 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 Compute Engine par défaut 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 dans le projet
  • Éditeur de données BigQuery (roles/bigquery.dataEditor) sur le compte de service de VM Dataproc dans le projet
  • Administrateur des objets de l'espace de stockage (roles/storage.objectAdmin) sur le compte de service de la VM Dataproc dans le projet
• Interrogez 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 des 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 un metastore BigLake, procédez comme suit :

1. Créez un cluster Dataproc ou configurez-en un existant.
2. Connectez-vous à votre moteur logiciel Open Source préféré, 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 au metastore BigLake à l'aide de SparkSQL 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 BigLake Metastore.

Ce fichier est inclus par défaut dans la version 2.2 de l'image Dataproc. 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 BigLake Metastore.

Configurer un cluster Dataproc

Avant de vous connecter à BigLake Metastore, vous devez configurer un cluster Dataproc.

Pour ce faire, vous pouvez créer un cluster ou en utiliser un existant. Vous utiliserez ensuite ce cluster pour exécuter Spark SQL de manière interactive et gérer vos ressources BigLake Metastore.

• L'accès privé à Google (APG) 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 n'ont 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 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 Zeppelin facultatif 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 au metastore BigLake à l'aide du client Flink SQL.

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 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 de base du répertoire de l'entrepôt (dossier Cloud Storage dans lequel Flink crée des fichiers). Cette valeur commence par gs://.
   • PROJECT_ID : ID du 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 metastore BigLake. Vous pouvez exécuter des commandes Flink SQL.

Gérer les ressources BigLake Metastore

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 Flink SQL interactive 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, ce 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 une table de 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 ingé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 point 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 le job 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 du job de streaming, procédez comme suit :

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

      accéder aux clusters

   2. Sélectionnez votre cluster.

   3. Cliquez sur l'onglet Interfaces Web.

   4. Cliquez sur le lien Gestionnaire de ressources YARN.

   5. Dans l'interface YARN ResourceManager, recherchez votre session Flink et cliquez sur le lien ApplicationMaster sous Tracking UI.

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

7. Interrogez les données de streaming dans le client Flink SQL :

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

8. Interroger les flux de données dans BigQuery :

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

9. Arrêtez le job de streaming dans le client Flink SQL :

   STOP JOB 'JOB_ID';

   Remplacez JOB_ID par l'ID de tâche qui s'est affiché dans le résultat lorsque vous avez créé la tâche de streaming.

Étapes suivantes

• Configurez les fonctionnalités BigLake Metastore facultatives.
• Affichez et interrogez des tables depuis Spark dans BigQuery.