Cette page explique comment créer un Dataproc un cluster exécutant Spark.
Présentation
Vous créez un cluster après l'association de l'instance de service Dataproc Metastore au lac Dataplex pour vous assurer que le cluster peut s'appuyer sur le point de terminaison du métastore Hive pour accéder aux métadonnées Dataplex.
Les métadonnées gérées dans Dataplex sont accessibles à l'aide d'interfaces standards, telles que Hive Metastore, pour alimenter les requêtes Spark. Les requêtes s'exécutent sur la cluster Dataproc.
Pour les données Parquet, définissez la propriété Spark spark.sql.hive.convertMetastoreParquet
sur false
pour éviter les erreurs d'exécution. En savoir plus
Créer un cluster Dataproc
Exécutez les commandes suivantes pour créer un cluster Dataproc, en spécifiant le service Dataproc Metastore associé au lac Dataplex :
GRPC_ENDPOINT=$(gcloud metastore services describe SERVICE_ID \
--location LOCATION \
--format "value(endpointUri)" | cut -c9-)
WHDIR=$(gcloud metastore services describe SERVICE_ID \
--location LOCATION \
--format "value(hiveMetastoreConfig.configOverrides.'hive.metastore.warehouse.dir')")
METASTORE_VERSION=$(gcloud metastore services describe SERVICE_ID \
--location LOCATION \
--format "value(hiveMetastoreConfig.version)")
# This command creates a cluster with default settings. You can customize
# it as needed. The --optional-components, --initialization-actions,
# --metadata and --properties flags are used to to connect with
# the associated metastore.
gcloud dataproc clusters create CLUSTER_ID \
--project PROJECT \
--region LOCATION \
--scopes "https://www.googleapis.com/auth/cloud-platform" \
--image-version 2.0-debian10 \
--optional-components=DOCKER \
--initialization-actions "gs://metastore-init-actions/metastore-grpc-proxy/metastore-grpc-proxy.sh" \
--metadata "proxy-uri=$GRPC_ENDPOINT,hive-version=$METASTORE_VERSION" \
--properties "hive:hive.metastore.uris=thrift://localhost:9083,hive:hive.metastore.warehouse.dir=$WHDIR"
Explorer les métadonnées
Exécutez des requêtes DQL pour explorer les métadonnées et des requêtes Spark pour interroger les données.
Avant de commencer
Ouvrez une session SSH sur le nœud principal du cluster Dataproc.
VM_ZONE=$(gcloud dataproc clusters describe CLUSTER_ID \ --project PROJECT \ --region LOCATION \ --format "value(config.gceClusterConfig.zoneUri)") gcloud compute ssh CLUSTER_ID-m --project PROJECT --zone $VM_ZONE
Lorsque l'invite de commande du nœud principal s'affiche, ouvrez un nouveau REPL Python.
python3
Répertorier des bases de données
Chaque zone Dataplex du lac est mappée sur une base de données Metastore.
import pyspark.sql as sql
session = sql.SparkSession.builder.enableHiveSupport().getOrCreate()
df = session.sql("SHOW DATABASES")
df.show()
Répertorier des tables
Répertorier les tables de l'une des zones.
import pyspark.sql as sql
session = sql.SparkSession.builder.enableHiveSupport().getOrCreate()
df = session.sql("SHOW TABLES IN ZONE_ID")
df.show()
Interroger les données
Interrogez les données de l'une des tables.
import pyspark.sql as sql
session = sql.SparkSession.builder.enableHiveSupport().getOrCreate()
# Modify the SQL statement to retrieve or filter on table columns.
df = session.sql("SELECT COLUMNS FROM ZONE_ID.TABLE_ID WHERE QUERY LIMIT 10")
df.show()
Créer des tables et des partitions dans les métadonnées
Exécutez des requêtes DDL pour créer des tables et des partitions dans les métadonnées Dataplex à l'aide d'Apache Spark.
Pour en savoir plus sur les types de données, les formats de fichiers et les formats de lignes compatibles, consultez Valeurs acceptées.
Avant de commencer
Avant de créer une table, créez un composant Dataplex qui se mappe sur le bucket Cloud Storage contenant les données sous-jacentes. Pour en savoir plus, consultez Ajouter un composant.
Créer une table
Les tables Parquet, ORC, AVRO, CSV et JSON sont acceptées.
import pyspark.sql as sql
session = sql.SparkSession.builder.enableHiveSupport().getOrCreate()
df = session.sql("CREATE TABLE ZONE_ID.TABLE_ID (COLUMNS DATA_TYPE) PARTITIONED BY (COLUMN) STORED AS FILE_FORMAT ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' LOCATION 'gs://MY_GCP_BUCKET/TABLE_LOCATION' TBLPROPERTIES('dataplex.entity.partition_style' = 'HIVE_COMPATIBLE')")
df.show()
Modifier une table
Dataplex ne vous permet pas de modifier l'emplacement d'une table ni la
colonnes de partition d'une table. Modifier une table ne définit pas automatiquement userManaged sur true
.
Dans Spark SQL, vous pouvez renommer une table, ajouter des colonnes, et définir le format de fichier d'une table.
Renommer une table
import pyspark.sql as sql
session = sql.SparkSession.builder.enableHiveSupport().getOrCreate()
df = session.sql("ALTER TABLE OLD_TABLE_NAME RENAME TO NEW_TABLE_NAME")
df.show()
Ajouter des colonnes
import pyspark.sql as sql
session = sql.SparkSession.builder.enableHiveSupport().getOrCreate()
df = session.sql("ALTER TABLE TABLE_NAME ADD COLUMN (COLUMN_NAME DATA_TYPE"))
df.show()
Définir le format de fichier
import pyspark.sql as sql
session = sql.SparkSession.builder.enableHiveSupport().getOrCreate()
df = session.sql("ALTER TABLE TABLE_NAME SET FILEFORMAT FILE_FORMAT")
df.show()
Déposer une table
Supprimer une table de l'API de métadonnées de Dataplex ne supprime pas les données sous-jacentes dans Cloud Storage.
import pyspark.sql as sql
session = sql.SparkSession.builder.enableHiveSupport().getOrCreate()
df = session.sql("DROP TABLE ZONE_ID.TABLE_ID")
df.show()
Ajouter une partition
Dataplex ne permet pas de modifier une partition une fois qu'elle est créée. Toutefois, la partition peut être supprimée.
import pyspark.sql as sql
session = sql.SparkSession.builder.enableHiveSupport().getOrCreate()
df = session.sql("ALTER TABLE ZONE_ID.TABLE_ID ADD PARTITION (COLUMN1=VALUE1) PARTITION (COLUMN2=VALUE2)")
df.show()
Vous pouvez ajouter plusieurs partitions de la même clé de partition et de valeurs de partition différentes, comme indiqué dans l'exemple précédent.
Supprimer une partition
Pour supprimer une partition, exécutez la commande suivante:
import pyspark.sql as sql
session = sql.SparkSession.builder.enableHiveSupport().getOrCreate()
df = session.sql("ALTER TABLE ZONE_ID.TABLE_ID DROP PARTITION (COLUMN=VALUE)")
df.show()
Interroger des tables Iceberg
Vous pouvez interroger des tables Iceberg à l'aide d'Apache Spark.
Avant de commencer
Configurez une session Spark SQL avec Iceberg.
spark-sql --packages org.apache.iceberg:iceberg-spark-runtime-3.1_2.12:0.13.1 --conf
spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions --conf
spark.sql.catalog.spark_catalog=org.apache.iceberg.spark.SparkSessionCatalog --conf
spark.sql.catalog.spark_catalog.type=hive --conf
spark.sql.catalog.local=org.apache.iceberg.spark.SparkCatalog --conf
spark.sql.catalog.local.type=hadoop --conf
spark.sql.catalog.local.warehouse=$PWD/warehouse
Créer une table Iceberg
Pour créer un tableau Iceberg, exécutez la commande suivante :
CREATE TABLE ZONE_ID.TABLE_ID (COLUMNS DATA_TYPE) USING ICEBERG PARTITIONED BY (COLUMN) LOCATION 'gs://MY_GCP_BUCKET/TABLE_ID' TBLPROPERTIES ('write.format.default' = 'TABLE_FORMAT');
Explorer l'instantané et l'historique Iceberg
Vous pouvez obtenir des instantanés et l'historique des tables Iceberg à l'aide d'Apache Spark.
Avant de commencer
Configurez une session PySpark compatible avec Iceberg.
pyspark --packages org.apache.iceberg:iceberg-spark-runtime-3.1_2.12:0.14.1 --conf
spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions --conf
spark.sql.catalog.spark_catalog=org.apache.iceberg.spark.SparkSessionCatalog --conf
spark.sql.catalog.spark_catalog.type=hive --conf
spark.sql.catalog.local=org.apache.iceberg.spark.SparkCatalog --conf
spark.sql.catalog.local.type=hadoop --conf
spark.sql.catalog.local.warehouse=$PWD/warehouse
Obtenir l'historique des tables d'Iceberg
Pour obtenir l'historique d'une table Iceberg, exécutez la commande suivante:
spark.read.format("iceberg").load("ZONE_ID.TABLE_ID.history").show(truncate=False)
Obtenir des instantanés de tables Iceberg
Pour obtenir un instantané d'une table Iceberg, exécutez la commande suivante :
spark.read.format("iceberg").load("ZONE_ID.TABLE_ID.snapshots").show(truncate=False, vertical=True)
Types de données et formats de fichiers compatibles
Les types de données acceptés sont définis comme suit:
Type de données | Valeurs |
---|---|
Primitif |
|
Tableau | ARRAY < DATA_TYPE > |
Structure | STRUCT < COLUMN : DATA_TYPE > |
Les formats de fichiers compatibles sont définis comme suit:
TEXTFILE
ORC
PARQUET
AVRO
JSONFILE
Pour en savoir plus sur les formats de fichiers, consultez la section Formats de stockage.
Les formats de ligne acceptés sont définis comme suit :
- DELIMITED [FIELDS TERMINATED BY CHAR]
- SERDE SERDE_NAME [AVEC SERDEPROPERTIES (PROPERTY_NAME=PROPERTY_VALUE, PROPERTY_NAME=PROPERTY_VALUE, ...)]
Étape suivante
- En savoir plus sur la gestion des métadonnées