Apache Kafka ist eine verteilte Open-Source-Streamingplattform für Echtzeit-Datenpipelines und Datenintegration. Es bietet ein effizientes und skalierbares Streamingsystem für den Einsatz in einer Vielzahl von Anwendungen, darunter:
- Echtzeitanalysen
- Streamverarbeitung
- Logaggregation
- Verteiltes Messaging
- Ereignisstreaming
Lernziele
Installieren Sie Kafka mit ZooKeeper auf einem Dataproc-HA-Cluster (in dieser Anleitung als "Dataproc-Kafka-Cluster" bezeichnet).
Sie erstellen fiktive Kundendaten und veröffentlichen die Daten dann in einem Kafka-Thema.
Erstellen Sie Hive-Parkett- und ORC-Tabellen in Cloud Storage, um gestreamte Kafka-Themendaten zu empfangen.
Senden Sie einen PySpark-Job, um das Kafka-Thema zu abonnieren und im Parquet- und ORC-Format in Cloud Storage zu streamen.
Führen Sie eine Abfrage für die gestreamten Hive-Tabellendaten aus, um die gestreamten Kafka-Nachrichten zu zählen.
Kosten
In diesem Dokument verwenden Sie die folgenden kostenpflichtigen Komponenten von Google Cloud:
Mit dem Preisrechner können Sie eine Kostenschätzung für Ihre voraussichtliche Nutzung vornehmen.
Nach Abschluss der in diesem Dokument beschriebenen Aufgaben können Sie weitere Kosten vermeiden, indem Sie die erstellten Ressourcen löschen. Weitere Informationen finden Sie unter Bereinigen.
Hinweise
Erstellen Sie ein Google Cloud-Projekt, falls noch nicht geschehen.
- Melden Sie sich bei Ihrem Google Cloud-Konto an. Wenn Sie mit Google Cloud noch nicht vertraut sind, erstellen Sie ein Konto, um die Leistungsfähigkeit unserer Produkte in der Praxis sehen und bewerten zu können. Neukunden erhalten außerdem ein Guthaben von 300 $, um Arbeitslasten auszuführen, zu testen und bereitzustellen.
-
In the Google Cloud console, on the project selector page, select or create a Google Cloud project.
-
Die Abrechnung für das Google Cloud-Projekt muss aktiviert sein.
-
Dataproc, Compute Engine, and Cloud Storage APIs aktivieren.
-
In the Google Cloud console, on the project selector page, select or create a Google Cloud project.
-
Die Abrechnung für das Google Cloud-Projekt muss aktiviert sein.
-
Dataproc, Compute Engine, and Cloud Storage APIs aktivieren.
- Wechseln Sie in der Google Cloud Console zur Cloud Storage-Seite Buckets.
- Klicken Sie auf Bucket erstellen.
- Geben Sie auf der Seite Bucket erstellen die Bucket-Informationen ein. Klicken Sie auf Weiter, um mit dem nächsten Schritt fortzufahren.
- Geben Sie unter Bucket benennen einen Namen ein, der den Anforderungen für Bucket-Namen entspricht.
-
Gehen Sie unter Speicherort für Daten auswählen folgendermaßen vor:
- Wählen Sie eine Option für Standorttyp aus.
- Wählen Sie eine Standort-Option aus.
- Wählen Sie unter Standardspeicherklasse für Ihre Daten auswählen eine Speicherklasse aus.
- Wählen Sie unter Zugriffssteuerung für Objekte auswählen eine Option für die Zugriffssteuerung aus.
- Geben Sie für Erweiterte Einstellungen (optional) eine Verschlüsselungsmethode, eine Aufbewahrungsrichtlinie oder Bucket-Labels an.
- Klicken Sie auf Erstellen.
Anleitungsschritte
Führen Sie die folgenden Schritte aus, um einen Kafka-Dataproc-Cluster zu erstellen, um ein Kafka-Thema im Parquet-ORC-Format in Cloud Storage zu lesen.
Kafka-Installationsskript in Cloud Storage kopieren
Das Skript kafka.sh
Initialisierungsaktion installiert Kafka auf einem Dataproc-Cluster.
Sehen Sie sich den Code an.
Kopieren Sie das Skript
kafka.sh
der Initialisierungsaktion in Ihren Cloud Storage-Bucket. Dieses Skript installiert Kafka auf einem Dataproc-Cluster.Öffnen Sie Cloud Shell und führen Sie den folgenden Befehl aus:
gsutil cp gs://goog-dataproc-initialization-actions-REGION/kafka/kafka.sh gs://BUCKET_NAME/scripts/
Ersetzen Sie die folgenden Werte:
- REGION:
kafka.sh
wird in öffentlichen Buckets mit regionalen Tags in Cloud Storage gespeichert. Geben Sie eine Compute Engine-Region an, die geografisch in der Nähe liegt (Beispiel:us-central1
). - BUCKET_NAME: Der Name Ihres Cloud Storage-Buckets
- REGION:
Dataproc-Kafka-Cluster erstellen
Öffnen Sie Cloud Shell und führen Sie dann den folgenden
gcloud dataproc clusters create
-Befehl aus, um einen Dataproc-HA-Cluster-Cluster zu erstellen, der die Kafka- und ZooKeeper-Komponenten installiert: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
Hinweise:
- KAFKA_CLUSTER: Der Clustername, der innerhalb eines Projekts eindeutig sein muss. Der Name muss mit einem Kleinbuchstaben beginnen und kann bis zu 51 Kleinbuchstaben, Ziffern und Bindestriche enthalten. Er darf nicht mit einem Bindestrich enden. Der Name eines gelöschten Clusters kann wiederverwendet werden.
- PROJECT_ID: Das Projekt, das mit diesem Cluster verknüpft werden soll.
- REGION: Die Compute Engine-Region, in der sich der Cluster befinden wird, z. B.
us-central1
.- Sie können das optionale Flag
--zone=ZONE
hinzufügen, um eine Zone innerhalb der angegebenen Region anzugeben, z. B.us-central1-a
. Wenn Sie keine Zone angeben, wählt das Dataproc-Feature für die automatische Zonenplatzierung eine Zone mit der angegebenen Region aus.
- Sie können das optionale Flag
--image-version
: Für diese Anleitung wird die Dataproc-Image-Version2.1-debian11
empfohlen. Hinweis: Jede Image-Version enthält eine Reihe vorinstallierter Komponenten, einschließlich der in dieser Anleitung verwendeten Hive-Komponente (siehe Unterstützte Dataproc-Image-Versionen).--num-master
:3
Masterknoten erstellen einen Hochverfügbarkeitscluster. Die für Kafka erforderliche Zookeeper-Komponente ist in einem Hochverfügbarkeitscluster vorinstalliert.--enable-component-gateway
: Aktiviert das Dataproc-Komponentengateway.- BUCKET_NAME: Der Name des Cloud Storage-Bucket, der das Initialisierungsskript
/scripts/kafka.sh
enthält (siehe Kafka-Installationsskript in Cloud Storage kopieren).
Kafka-custdata
-Thema erstellen
So erstellen Sie ein Kafka-Thema im Dataproc-Kafka-Cluster:
Öffnen Sie mit dem SSH-Dienstprogramm ein Terminalfenster auf der Clustermaster-VM.
Erstellen Sie ein Kafka-
custdata
-Thema./usr/lib/kafka/bin/kafka-topics.sh \ --bootstrap-server KAFKA_CLUSTER-w-0:9092 \ --create --topic custdata
Hinweise:
KAFKA_CLUSTER: Geben Sie den Namen Ihres Kafka-Clusters ein.
-w-0:9092
gibt den Kafka-Broker an, der an Port9092
auf dem Knotenworker-0
ausgeführt wird.Nachdem Sie das Thema
custdata
erstellt haben, können Sie die folgenden Befehle ausführen:# 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
Inhalte im Kafka-Thema custdata
veröffentlichen
Das folgende Skript verwendet das Kafka-Tool kafka-console-producer.sh
, um fiktive Kundendaten im CSV-Format zu generieren.
Kopieren Sie das Skript und fügen Sie es im SSH-Terminal auf dem Masterknoten Ihres Kafka-Clusters ein. Drücken Sie die <return>, um das Skript auszuführen.
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=:"
Hinweise:
- KAFKA_CLUSTER: Der Name Ihres Kafka-Clusters.
Führen Sie den folgenden Kafka-Befehl aus, um zu prüfen, ob das Thema
custdata
10.000 Nachrichten enthält./usr/lib/kafka/bin/kafka-run-class.sh kafka.tools.GetOffsetShell \ --broker-list KAFKA_CLUSTER-w-0:9092 \ --topic custdata
Hinweise:
- KAFKA_CLUSTER: Der Name Ihres Kafka-Clusters.
Erwartete Ausgabe:
custdata:0:10000
Hive-Tabellen in Cloud Storage erstellen
Hive-Tabellen erstellen, um gestreamte Kafka-Themendaten zu empfangen.
Führen Sie die folgenden Schritte aus, um cust_parquet
(parquet) und eine cust_orc
(ORC) Hive-Tabelle in Ihrem Cloud Storage-Bucket zu erstellen.
Fügen Sie BUCKET_NAME in das folgende Skript ein. Kopieren Sie es anschließend in das SSH-Terminal Ihres Kafka-Cluster-Masterknotens und drücken Sie <return>, um ein
~/hivetables.hql
(Hive Query Language)-Skript zu erstellen.Im nächsten Schritt führen Sie das Skript
~/hivetables.hql
aus, um Parquet- und ORC Hive-Tabellen in Ihrem Cloud Storage-Bucket zu erstellen.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
Senden Sie im SSH-Terminal auf dem Masterknoten Ihres Kafka-Clusters den Hive-Job
~/hivetables.hql
, um im Cloud Storage-Bucketcust_parquet
(Parquet) undcust_orc
(ORC) Hive-Tabellen zu erstellen.gcloud dataproc jobs submit hive \ --cluster=KAFKA_CLUSTER \ --region=REGION \ -f ~/hivetables.hql
Hinweise:
- Die Hive-Komponente ist auf dem Kafka-Cluster von Dataproc vorinstalliert. Eine Liste der Hive-Komponentenversionen, die in kürzlich veröffentlichten 2.1-Images enthalten sind, finden Sie unter Releaseversionen 2.1.x.
- KAFKA_CLUSTER: Der Name Ihres Kafka-Clusters.
- REGION: Die Region, in der sich Ihr Kafka-Cluster befindet.
Kafka-custdata
zu Hive-Tabellen streamen
- Führen Sie den folgenden Befehl im SSH-Terminal auf dem Masterknoten Ihres Kafka-Clusters aus, um die
kafka-python
-Bibliothek zu installieren. Zum Streamen von Kafka-Themendaten in Cloud Storage ist ein Kafka-Client erforderlich.
pip install kafka-python
Fügen Sie BUCKET_NAME ein. Kopieren Sie dann den folgenden PySpark-Code und fügen Sie ihn in das SSH-Terminal Ihres Kafka-Cluster-Masterknotens ein. Drücken Sie dann <return>, um eine
streamdata.py
-Datei zu erstellen.Das Skript abonniert das Kafka-Thema
custdata
und streamt die Daten dann in Ihre Hive-Tabellen in Cloud Storage. Das Ausgabeformat, bei dem es sich um Parquet oder ORC handeln kann, wird als Parameter an das Skript übergeben.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
Führen Sie im SSH-Terminal auf dem Masterknoten Ihres Kafka-Clusters
spark-submit
aus, um Daten in Ihre Hive-Tabellen in Cloud Storage zu streamen.Fügen Sie den Namen Ihres KAFKA_CLUSTER und der Ausgabe-FORMAT ein. Kopieren Sie dann den folgenden Code und fügen Sie ihn in das SSH-Terminal auf dem Masterknoten Ihres Kafka-Clusters ein. Drücken Sie dann <return>, um den Code auszuführen und die Kafka-
custdata
-Daten im Parquet-Format in Ihre Hive-Tabellen in Cloud Storage zu streamen.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
Hinweise:
- KAFKA_CLUSTER: Geben Sie den Namen Ihres Kafka-Clusters ein.
- FORMAT: Geben Sie entweder
parquet
oderorc
als Ausgabeformat an. Sie können den Befehl nacheinander ausführen, um beide Formate in die Hive-Tabellen zu streamen. Geben Sie beispielsweise beim ersten Aufrufparquet
an, um das Kafka-custdata
-Thema in die Hive-Parquet-Tabelle zu streamen. Geben Sie dann im zweiten Aufruf das Formatorc
an, umcustdata
in die Hive-ORC-Tabelle zu streamen.
Nachdem die Standardausgabe im SSH-Terminal angehalten wurde, was bedeutet, dass alle
custdata
gestreamt wurden, drücken Sie im SSH-Terminal <control-c>, um den Prozess zu beenden.Listen Sie die Hive-Tabellen in Cloud Storage auf.
gsutil ls -r gs://BUCKET_NAME/tables/*
Hinweise:
- BUCKET_NAME: Geben Sie den Namen des Cloud Storage-Buckets ein, der Ihre Hive-Tabellen enthält (siehe Hive-Tabellen erstellen).
Streamingdaten abfragen
Führen Sie im SSH-Terminal auf dem Masterknoten Ihres Kafka-Clusters den folgenden
hive
-Befehl aus, um die gestreamten Kafka-custdata
-Nachrichten in den Hive-Tabellen in Cloud Storage zu zählen.hive -e "select count(1) from TABLE_NAME"
Hinweise:
- TABLE_NAME: Geben Sie entweder
cust_parquet
odercust_orc
als Namen der Hive-Tabelle an.
Erwartetes Ausgabe-Snippet:
- TABLE_NAME: Geben Sie entweder
...
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)
Bereinigen
Projekt löschen
Google Cloud-Projekt löschen:
gcloud projects delete PROJECT_ID
Ressourcen löschen
- Löschen Sie den Bucket:
gcloud storage buckets delete BUCKET_NAME
- Löschen Sie den Kafka-Cluster:
gcloud dataproc clusters delete KAFKA_CLUSTER \ --region=${REGION}