Apache Kafka est une plate-forme de streaming distribué Open Source pour les pipelines de données en temps réel et l'intégration de données. Il fournit un système de streaming efficace et évolutif pour une utilisation dans diverses applications, y compris:
- Analyse en temps réel
- Traitement par flux
- Agrégation de journaux
- Messagerie distribuée
- Streaming d'événements
Objectifs
Installez Kafka sur un cluster Dataproc HA avec ZooKeeper (appelé "cluster Dataproc Kafka" dans ce tutoriel).
Créer des données client fictives, puis les publier dans un sujet Kafka.
Créer du parquet Hive et des tables ORC dans Cloud Storage pour recevoir des données de sujet Kafka diffusées en flux continu.
Envoyer une tâche PySpark pour s'abonner au sujet Kafka et le diffuser dans Cloud Storage au format Parquet et ORC
Exécutez une requête sur les données de la table Hive diffusées en flux continu pour compter les messages Kafka diffusés.
Coûts
Dans ce document, vous utilisez les composants facturables suivants de Google Cloud :
Obtenez une estimation des coûts en fonction de votre utilisation prévue à l'aide du simulateur de coût.
Une fois que vous avez terminé les tâches décrites dans ce document, vous pouvez éviter de continuer à payer des frais en supprimant les ressources que vous avez créées. Pour en savoir plus, consultez la section Effectuer un nettoyage.
Avant de commencer
Si vous ne l'avez pas déjà fait, créez un projet Google Cloud.
- 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 gratuits 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.
-
Vérifiez que la facturation est activée pour votre projet Google Cloud.
-
Activer les API Dataproc, Compute Engine, and Cloud Storage.
-
In the Google Cloud console, on the project selector page, select or create a Google Cloud project.
-
Vérifiez que la facturation est activée pour votre projet Google Cloud.
-
Activer les API Dataproc, Compute Engine, and Cloud Storage.
- Dans la console Google Cloud, accédez à la page Buckets Cloud Storage.
- Cliquez sur Créer un bucket.
- Sur la page Créer un bucket, saisissez les informations concernant votre bucket. Pour passer à l'étape suivante, cliquez sur Continuer.
- Pour nommer votre bucket, saisissez un nom qui répond aux exigences de dénomination des buckets.
-
Pour Choisir l'emplacement de stockage des données, procédez comme suit :
- Sélectionnez une option de type d'emplacement.
- Sélectionnez une option Location (Emplacement).
- Pour Choisir une classe de stockage par défaut pour vos données, sélectionnez une classe de stockage.
- Pour le champ Choisir comment contrôler l'accès aux objets, sélectionnez une option de Contrôle des accès.
- Sous Paramètres avancés (facultatif), choisissez une méthode de chiffrement, une règle de conservation ou des libellés de bucket.
- Cliquez sur Create (Créer).
Étapes du tutoriel
Procédez comme suit pour créer un cluster Dataproc Kafka afin de lire un sujet Kafka dans Cloud Storage au format Parquet OU ORC.
Copier le script d'installation de Kafka dans Cloud Storage
Le script d'action d'initialisation kafka.sh
installe Kafka sur un cluster Dataproc.
Parcourez le code.
Copiez le script d'action d'initialisation
kafka.sh
dans votre bucket Cloud Storage. Ce script installe Kafka sur un cluster Dataproc.Ouvrez Cloud Shell, puis exécutez la commande suivante :
gsutil cp gs://goog-dataproc-initialization-actions-REGION/kafka/kafka.sh gs://BUCKET_NAME/scripts/
Effectuez les remplacements suivants :
- REGION:
kafka.sh
est stocké dans des buckets publics avec des tags régionaux dans Cloud Storage. Spécifiez une région Compute Engine proche géographiquement (exemple:us-central1
). - BUCKET_NAME : nom du bucket Cloud Storage.
- REGION:
Créer un cluster Dataproc Kafka
Ouvrez Cloud Shell, puis exécutez la commande
gcloud dataproc clusters create
suivante pour créer un cluster de cluster à haute disponibilité Dataproc qui installe les composants Kafka et ZooKeeper:gcloud dataproc clusters create KAFKA_CLUSTER \ --project=PROJECT_ID \ --region=REGION \ --image-version=2.1-debian11 \ --num-masters=3 \ --enable-component-gateway \ --initialization-actions=gs://BUCKET_NAME/scripts/kafka.sh
Remarques :
- KAFKA_CLUSTER: nom du cluster, qui doit être unique au sein d'un projet. Le nom doit commencer par une lettre minuscule et peut contenir jusqu'à 51 lettres minuscules, chiffres et traits d'union. Il ne peut pas se terminer par un trait d'union. Le nom d'un cluster supprimé peut être réutilisé.
- PROJECT_ID: projet à associer à ce cluster.
- REGION: région Compute Engine dans laquelle le cluster sera situé, par exemple
us-central1
.- Vous pouvez ajouter l'option facultative
--zone=ZONE
pour spécifier une zone dans la région spécifiée, par exempleus-central1-a
. Si vous ne spécifiez pas de zone, la fonctionnalité Dataproc autozone placement sélectionne une zone avec la région spécifiée.
- Vous pouvez ajouter l'option facultative
--image-version
: la version d'image Dataproc2.1-debian11
est recommandée pour ce tutoriel. Remarque: Chaque version d'image contient un ensemble de composants préinstallés, y compris le composant Hive utilisé dans ce tutoriel (consultez la section Versions d'image Dataproc compatibles).--num-master
: les nœuds maîtres3
créent un cluster à haute disponibilité. Le composant ZooKeeper, requis par Kafka, est préinstallé sur un cluster à haute disponibilité.--enable-component-gateway
: active la passerelle des composants Dataproc.- BUCKET_NAME: nom du bucket Cloud Storage contenant le script d'initialisation
/scripts/kafka.sh
(consultez la section Copier le script d'installation de Kafka dans Cloud Storage).
Créer un sujet custdata
Kafka
Pour créer un sujet Kafka sur le cluster Dataproc Kafka:
Utilisez l'utilitaire SSH pour ouvrir une fenêtre de terminal sur la VM maître du cluster.
Créez un sujet
custdata
Kafka./usr/lib/kafka/bin/kafka-topics.sh \ --bootstrap-server KAFKA_CLUSTER-w-0:9092 \ --create --topic custdata
Remarques :
KAFKA_CLUSTER: insérez le nom de votre cluster Kafka.
-w-0:9092
signifie l'agent Kafka s'exécutant sur le port9092
du nœudworker-0
.Vous pouvez exécuter les commandes suivantes après avoir créé le sujet
custdata
:# List all topics. /usr/lib/kafka/bin/kafka-topics.sh \ --bootstrap-server KAFKA_CLUSTER-w-0:9092 \ --list
# Consume then display topic data. /usr/lib/kafka/bin/kafka-console-consumer.sh \ --bootstrap-server KAFKA_CLUSTER-w-0:9092 \ --topic custdata
# Count the number of messages in the topic. /usr/lib/kafka/bin/kafka-run-class.sh kafka.tools.GetOffsetShell \ --broker-list KAFKA_CLUSTER-w-0:9092 \ --topic custdata # Delete topic. /usr/lib/kafka/bin/kafka-topics.sh \ --bootstrap-server KAFKA_CLUSTER-w-0:9092 \ --delete --topic custdata
Publier du contenu sur le sujet custdata
Kafka
Le script suivant utilise l'outil Kafka kafka-console-producer.sh
pour générer des données client fictives au format CSV.
Copiez et collez le script dans le terminal SSH du nœud maître de votre cluster Kafka. Appuyez sur <return> pour exécuter le script.
for i in {1..10000}; do \ custname="cust name${i}" uuid=$(dbus-uuidgen) age=$((45 + $RANDOM % 45)) amount=$(echo "$(( $RANDOM % 99999 )).$(( $RANDOM % 99 ))") message="${uuid}:${custname},${age},${amount}" echo ${message} done | /usr/lib/kafka/bin/kafka-console-producer.sh \ --broker-list KAFKA_CLUSTER-w-0:9092 \ --topic custdata \ --property "parse.key=true" \ --property "key.separator=:"
Remarques :
- KAFKA_CLUSTER: nom de votre cluster Kafka.
Exécutez la commande Kafka suivante pour vérifier que le sujet
custdata
contient 10 000 messages./usr/lib/kafka/bin/kafka-run-class.sh kafka.tools.GetOffsetShell \ --broker-list KAFKA_CLUSTER-w-0:9092 \ --topic custdata
Remarques :
- KAFKA_CLUSTER: nom de votre cluster Kafka.
Résultat attendu :
custdata:0:10000
Créer des tables Hive dans Cloud Storage
Créer des tables Hive pour recevoir des données sur le sujet Kafka diffusées en flux continu
Procédez comme suit pour créer des tables Hive cust_parquet
(parquet) et cust_orc
(ORC) dans votre bucket Cloud Storage.
Insérez votre BUCKET_NAME dans le script suivant, copiez et collez le script dans le terminal SSH du nœud maître du cluster Kafka, puis appuyez sur <return> pour créer un script
~/hivetables.hql
(Hive Query Language).À l'étape suivante, vous allez exécuter le script
~/hivetables.hql
pour créer des tables Parquet et Hive ORC dans votre bucket Cloud Storage.cat > ~/hivetables.hql <<EOF drop table if exists cust_parquet; create external table if not exists cust_parquet (uuid string, custname string, age string, amount string) row format delimited fields terminated by ',' stored as parquet location "gs://BUCKET_NAME/tables/cust_parquet"; drop table if exists cust_orc; create external table if not exists cust_orc (uuid string, custname string, age string, amount string) row format delimited fields terminated by ',' stored as orc location "gs://BUCKET_NAME/tables/cust_orc"; EOF
Dans le terminal SSH du nœud maître de votre cluster Kafka, envoyez la tâche Hive
~/hivetables.hql
pour créer des tables Hivecust_parquet
(parquet) etcust_orc
(ORC) Hive dans votre bucket Cloud Storage.gcloud dataproc jobs submit hive \ --cluster=KAFKA_CLUSTER \ --region=REGION \ -f ~/hivetables.hql
Remarques :
- Le composant Hive est préinstallé sur le cluster Dataproc Kafka. Consultez la section Versions 2.1.x pour obtenir la liste des versions des composants Hive incluses dans les images 2.1 récemment publiées.
- KAFKA_CLUSTER: nom de votre cluster Kafka.
- REGION: région où se trouve votre cluster Kafka.
Diffuser des custdata
Kafka vers des tables Hive
- Exécutez la commande suivante dans le terminal SSH du nœud maître de votre cluster Kafka pour installer la bibliothèque
kafka-python
. Un client Kafka est nécessaire pour diffuser les données du sujet Kafka vers Cloud Storage.
pip install kafka-python
Insérez votre BUCKET_NAME, puis copiez et collez le code PySpark suivant dans le terminal SSH de votre nœud maître de cluster Kafka, puis appuyez sur <return> pour créer un fichier
streamdata.py
.Le script s'abonne au sujet
custdata
Kafka, puis diffuse les données vers vos tables Hive dans Cloud Storage. Le format de sortie, qui peut être Parquet ou ORC, est transmis au script en tant que paramètre.cat > streamdata.py <<EOF #!/bin/python import sys from pyspark.sql.functions import * from pyspark.sql.types import * from pyspark.sql import SparkSession from kafka import KafkaConsumer def getNameFn (data): return data.split(",")[0] def getAgeFn (data): return data.split(",")[1] def getAmtFn (data): return data.split(",")[2] def main(cluster, outputfmt): spark = SparkSession.builder.appName("APP").getOrCreate() spark.sparkContext.setLogLevel("WARN") Logger = spark._jvm.org.apache.log4j.Logger logger = Logger.getLogger(__name__) rows = spark.readStream.format("kafka") \ .option("kafka.bootstrap.servers", cluster+"-w-0:9092").option("subscribe", "custdata") \ .option("startingOffsets", "earliest")\ .load() getNameUDF = udf(getNameFn, StringType()) getAgeUDF = udf(getAgeFn, StringType()) getAmtUDF = udf(getAmtFn, StringType()) logger.warn("Params passed in are cluster name: " + cluster + " output format(sink): " + outputfmt) query = rows.select (col("key").cast("string").alias("uuid"),\ getNameUDF (col("value").cast("string")).alias("custname"),\ getAgeUDF (col("value").cast("string")).alias("age"),\ getAmtUDF (col("value").cast("string")).alias("amount")) writer = query.writeStream.format(outputfmt)\ .option("path","gs://BUCKET_NAME/tables/cust_"+outputfmt)\ .option("checkpointLocation", "gs://BUCKET_NAME/chkpt/"+outputfmt+"wr") \ .outputMode("append")\ .start() writer.awaitTermination() if __name__=="__main__": if len(sys.argv) < 2: print ("Invalid number of arguments passed ", len(sys.argv)) print ("Usage: ", sys.argv[0], " cluster format") print ("e.g.: ", sys.argv[0], " <cluster_name> orc") print ("e.g.: ", sys.argv[0], " <cluster_name> parquet") main(sys.argv[1], sys.argv[2]) EOF
Dans le terminal SSH du nœud maître de votre cluster Kafka, exécutez
spark-submit
pour diffuser des données en flux continu vers vos tables Hive dans Cloud Storage.Insérez le nom de votre KAFKA_CLUSTER et la sortie FORMAT, puis copiez et collez le code suivant dans le terminal SSH du nœud maître de votre cluster Kafka, puis appuyez sur <return> pour exécuter le code et diffuser les données
custdata
de Kafka au format Parquet vers vos tables Hive dans Cloud Storage.spark-submit --packages \ org.apache.spark:spark-streaming-kafka-0-10_2.12:3.1.3,org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.3 \ --conf spark.history.fs.gs.outputstream.type=FLUSHABLE_COMPOSITE \ --conf spark.driver.memory=4096m \ --conf spark.executor.cores=2 \ --conf spark.executor.instances=2 \ --conf spark.executor.memory=6144m \ streamdata.py KAFKA_CLUSTER FORMAT
Remarques :
- KAFKA_CLUSTER: insérez le nom de votre cluster Kafka.
- FORMAT: spécifiez
parquet
ouorc
comme format de sortie. Vous pouvez exécuter la commande successivement pour diffuser les deux formats vers les tables Hive: par exemple, dans le premier appel, spécifiezparquet
pour diffuser le sujet Kafkacustdata
vers la table parquet Hive ; puis, lors du deuxième appel, spécifiez le formatorc
pour diffusercustdata
vers la table ORC Hive.
Une fois que la sortie standard s'est arrêtée dans le terminal SSH, ce qui signifie que toutes les
custdata
ont été diffusées, appuyez sur <control-c> dans le terminal SSH pour arrêter le processus.Lister les tables Hive dans Cloud Storage
gsutil ls -r gs://BUCKET_NAME/tables/*
Remarques :
- BUCKET_NAME: insérez le nom du bucket Cloud Storage contenant vos tables Hive (consultez Créer des tables Hive).
Interroger les données diffusées
Dans le terminal SSH du nœud maître de votre cluster Kafka, exécutez la commande
hive
suivante pour compter les messagescustdata
Kafka diffusés dans les tables Hive de Cloud Storage.hive -e "select count(1) from TABLE_NAME"
Remarques :
- TABLE_NAME: spécifiez
cust_parquet
oucust_orc
comme nom de la table Hive.
Extrait de résultat attendu:
- TABLE_NAME: spécifiez
...
Status: Running (Executing on YARN cluster with App id application_....)
----------------------------------------------------------------------------------------------
VERTICES MODE STATUS TOTAL COMPLETED RUNNING PENDING FAILED KILLED
----------------------------------------------------------------------------------------------
Map 1 .......... container SUCCEEDED 1 1 0 0 0 0
Reducer 2 ...... container SUCCEEDED 1 1 0 0 0 0
----------------------------------------------------------------------------------------------
VERTICES: 02/02 [==========================>>] 100% ELAPSED TIME: 9.89 s
----------------------------------------------------------------------------------------------
OK
10000
Time taken: 21.394 seconds, Fetched: 1 row(s)
Effectuer un nettoyage
Supprimer le projet
Supprimez un projet Google Cloud :
gcloud projects delete PROJECT_ID
Delete resources
-
Supprimez le bucket :
gcloud storage buckets delete BUCKET_NAME
- Supprimez votre cluster Kafka :
gcloud dataproc clusters delete KAFKA_CLUSTER \ --region=${REGION}