Kafka-Thema in Hive streamen

---

Apache Kafka ist eine verteilte Open-Source-Streamingplattform für Echtzeitdaten Pipelines und Datenintegration. Es bietet ein effizientes und skalierbares Streaming-System. zur Verwendung in einer Vielzahl von Anwendungen, einschließlich:

• Echtzeitanalysen
• Streamverarbeitung
• Logaggregation
• Verteiltes Messaging
• Ereignisstreaming

Weitere Informationen finden Sie unter Kafka-Anwendungsfälle.

Lernziele

1. Installieren Sie Kafka auf einem Dataproc-HA-Cluster mit ZooKeeper (in dieser Anleitung als "Dataproc-Kafka-Cluster" bezeichnet).

2. Sie erstellen fiktive Kundendaten und veröffentlichen die Daten dann in einem Kafka-Thema.

3. Hive-Parkett- und ORC-Tabellen in Cloud Storage erstellen um gestreamte Kafka-Themendaten zu erhalten.

4. Senden Sie einen PySpark-Job, um das Kafka-Thema zu abonnieren und in dieses zu streamen Cloud Storage im Parquet- und ORC-Format.

5. Abfrage der gestreamten Hive-Tabellendaten ausführen, um die gestreamten Daten zu zählen Kafka-Nachrichten.

Kosten

In diesem Dokument verwenden Sie die folgenden kostenpflichtigen Komponenten von Google Cloud:

• Dataproc
• Compute Engine
• Cloud Storage

Mit dem Preisrechner können Sie eine Kostenschätzung für Ihre voraussichtliche Nutzung vornehmen. Neuen Google Cloud-Nutzern steht möglicherweise eine kostenlose Testversion zur Verfügung.

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.

1. 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.

2. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

3. Die Abrechnung für das Google Cloud-Projekt muss aktiviert sein.

4. Dataproc, Compute Engine, and Cloud Storage APIs aktivieren.

   Aktivieren Sie die APIs

5. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

6. Die Abrechnung für das Google Cloud-Projekt muss aktiviert sein.

7. Dataproc, Compute Engine, and Cloud Storage APIs aktivieren.

   Aktivieren Sie die APIs

8. Wechseln Sie in der Google Cloud Console zur Cloud Storage-Seite Buckets.

   Zur Seite „Buckets“

9. Klicken Sie auf Bucket erstellen.
10. 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.
11. Klicken Sie auf Erstellen.

Anleitungsschritte

Führen Sie die folgenden Schritte aus, um einen Dataproc-Kafka-Cluster zu erstellen, Ein Kafka-Thema im Parquet-ORC-Format in Cloud Storage einlesen.

Hinweis: In einer Produktionsumgebung werden einige der Aufgaben, die in der werden die folgenden Abschnitte in separaten Clustern ausgeführt. Für bestimmte Zwecke In dieser Anleitung werden alle Aufgaben auf einem einzigen Dataproc ausgeführt. Kafka-Cluster.

Kafka-Installationsskript in Cloud Storage kopieren

Die Initialisierungsaktion kafka.sh Das Skript installiert Kafka auf einem Dataproc-Cluster.

1. Sehen Sie sich den Code an.
   

   <ph type="x-smartling-placeholder"></ph> Maximieren Minimieren

   #!/bin/bash
   #    Copyright 2015 Google, Inc.
   #
   #    Licensed under the Apache License, Version 2.0 (the "License");
   #    you may not use this file except in compliance with the License.
   #    You may obtain a copy of the License at
   #
   #        http://www.apache.org/licenses/LICENSE-2.0
   #
   #    Unless required by applicable law or agreed to in writing, software
   #    distributed under the License is distributed on an "AS IS" BASIS,
   #    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   #    See the License for the specific language governing permissions and
   #    limitations under the License.
   #
   # This script installs Apache Kafka (http://kafka.apache.org) on a Google Cloud
   # Dataproc cluster.
   
   set -euxo pipefail
   
   readonly ZOOKEEPER_HOME=/usr/lib/zookeeper
   readonly KAFKA_HOME=/usr/lib/kafka
   readonly KAFKA_PROP_FILE='/etc/kafka/conf/server.properties'
   readonly ROLE="$(/usr/share/google/get_metadata_value attributes/dataproc-role)"
   readonly RUN_ON_MASTER="$(/usr/share/google/get_metadata_value attributes/run-on-master || echo false)"
   readonly KAFKA_ENABLE_JMX="$(/usr/share/google/get_metadata_value attributes/kafka-enable-jmx || echo false)"
   readonly KAFKA_JMX_PORT="$(/usr/share/google/get_metadata_value attributes/kafka-jmx-port || echo 9999)"
   readonly INSTALL_KAFKA_PYTHON="$(/usr/share/google/get_metadata_value attributes/install-kafka-python || echo false)"
   
   # The first ZooKeeper server address, e.g., "cluster1-m-0:2181".
   ZOOKEEPER_ADDRESS=''
   # Integer broker ID of this node, e.g., 0
   BROKER_ID=''
   
   function retry_apt_command() {
     cmd="$1"
     for ((i = 0; i < 10; i++)); do
       if eval "$cmd"; then
         return 0
       fi
       sleep 5
     done
     return 1
   }
   
   function recv_keys() {
     retry_apt_command "apt-get install -y gnupg2 &&\
                        apt-key adv --keyserver keyserver.ubuntu.com --recv-keys B7B3B788A8D3785C"
   }
   
   function update_apt_get() {
     retry_apt_command "apt-get update"
   }
   
   function install_apt_get() {
     pkgs="$@"
     retry_apt_command "apt-get install -y $pkgs"
   }
   
   function err() {
     echo "[$(date +'%Y-%m-%dT%H:%M:%S%z')]: $@" >&2
     return 1
   }
   
   # Returns the list of broker IDs registered in ZooKeeper, e.g., " 0, 2, 1,".
   function get_broker_list() {
     ${KAFKA_HOME}/bin/zookeeper-shell.sh "${ZOOKEEPER_ADDRESS}" \
       <<<"ls /brokers/ids" |
       grep '\[.*\]' |
       sed 's/\[/ /' |
       sed 's/\]/,/'
   }
   
   # Waits for zookeeper to be up or time out.
   function wait_for_zookeeper() {
     for i in {1..20}; do
       if "${ZOOKEEPER_HOME}/bin/zkCli.sh" -server "${ZOOKEEPER_ADDRESS}" ls /; then
         return 0
       else
         echo "Failed to connect to ZooKeeper ${ZOOKEEPER_ADDRESS}, retry ${i}..."
         sleep 5
       fi
     done
     echo "Failed to connect to ZooKeeper ${ZOOKEEPER_ADDRESS}" >&2
     exit 1
   }
   
   # Wait until the current broker is registered or time out.
   function wait_for_kafka() {
     for i in {1..20}; do
       local broker_list=$(get_broker_list || true)
       if [[ "${broker_list}" == *" ${BROKER_ID},"* ]]; then
         return 0
       else
         echo "Kafka broker ${BROKER_ID} is not registered yet, retry ${i}..."
         sleep 5
       fi
     done
     echo "Failed to start Kafka broker ${BROKER_ID}." >&2
     exit 1
   }
   
   function install_and_configure_kafka_server() {
     # Find zookeeper list first, before attempting any installation.
     local zookeeper_client_port
     zookeeper_client_port=$(grep 'clientPort' /etc/zookeeper/conf/zoo.cfg |
       tail -n 1 |
       cut -d '=' -f 2)
   
     local zookeeper_list
     zookeeper_list=$(grep '^server\.' /etc/zookeeper/conf/zoo.cfg |
       cut -d '=' -f 2 |
       cut -d ':' -f 1 |
       sort |
       uniq |
       sed "s/$/:${zookeeper_client_port}/" |
       xargs echo |
       sed "s/ /,/g")
   
     if [[ -z "${zookeeper_list}" ]]; then
       # Didn't find zookeeper quorum in zoo.cfg, but possibly workers just didn't
       # bother to populate it. Check if YARN HA is configured.
       zookeeper_list=$(bdconfig get_property_value --configuration_file \
         /etc/hadoop/conf/yarn-site.xml \
         --name yarn.resourcemanager.zk-address 2>/dev/null)
     fi
   
     # If all attempts failed, error out.
     if [[ -z "${zookeeper_list}" ]]; then
       err 'Failed to find configured Zookeeper list; try "--num-masters=3" for HA'
     fi
   
     ZOOKEEPER_ADDRESS="${zookeeper_list%%,*}"
   
     # Install Kafka from Dataproc distro.
     install_apt_get kafka-server || dpkg -l kafka-server ||
       err 'Unable to install and find kafka-server.'
   
     mkdir -p /var/lib/kafka-logs
     chown kafka:kafka -R /var/lib/kafka-logs
   
     if [[ "${ROLE}" == "Master" ]]; then
       # For master nodes, broker ID starts from 10,000.
       if [[ "$(hostname)" == *-m ]]; then
         # non-HA
         BROKER_ID=10000
       else
         # HA
         BROKER_ID=$((10000 + $(hostname | sed 's/.*-m-\([0-9]*\)$/\1/g')))
       fi
     else
       # For worker nodes, broker ID is the worker ID.
       BROKER_ID=$(hostname | sed 's/.*-w-\([0-9]*\)$/\1/g')
     fi
     sed -i 's|log.dirs=/tmp/kafka-logs|log.dirs=/var/lib/kafka-logs|' \
       "${KAFKA_PROP_FILE}"
     sed -i 's|^\(zookeeper\.connect=\).*|\1'${zookeeper_list}'|' \
       "${KAFKA_PROP_FILE}"
     sed -i 's,^\(broker\.id=\).*,\1'${BROKER_ID}',' \
       "${KAFKA_PROP_FILE}"
     echo -e '\nreserved.broker.max.id=100000' >>"${KAFKA_PROP_FILE}"
     echo -e '\ndelete.topic.enable=true' >>"${KAFKA_PROP_FILE}"
   
     if [[ "${KAFKA_ENABLE_JMX}" == "true" ]]; then
       sed -i '/kafka-run-class.sh/i export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote=true -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname=localhost -Djava.net.preferIPv4Stack=true"' /usr/lib/kafka/bin/kafka-server-start.sh
       sed -i "/kafka-run-class.sh/i export JMX_PORT=${KAFKA_JMX_PORT}" /usr/lib/kafka/bin/kafka-server-start.sh
     fi
   
     wait_for_zookeeper
   
     # Start Kafka.
     service kafka-server restart
   
     wait_for_kafka
   }
   
   function install_kafka_python_package() {
     KAFKA_PYTHON_PACKAGE="kafka-python==2.0.2"
     if [[ "${INSTALL_KAFKA_PYTHON}" != "true" ]]; then
       return
     fi
   
     if [[ "$(echo "${DATAPROC_IMAGE_VERSION} > 2.0" | bc)" -eq 1 ]]; then
       /opt/conda/default/bin/pip install "${KAFKA_PYTHON_PACKAGE}" || { sleep 10; /opt/conda/default/bin/pip install "${KAFKA_PYTHON_PACKAGE}"; }
     else
       OS=$(. /etc/os-release && echo "${ID}")
       if [[ "${OS}" == "rocky" ]]; then
         yum install -y python2-pip
       else
         apt-get install -y python-pip
       fi
       pip2 install "${KAFKA_PYTHON_PACKAGE}" || { sleep 10; pip2 install "${KAFKA_PYTHON_PACKAGE}"; } || { sleep 10; pip install "${KAFKA_PYTHON_PACKAGE}"; }
     fi
   }
   
   function remove_old_backports {
     # This script uses 'apt-get update' and is therefore potentially dependent on
     # backports repositories which have been archived.  In order to mitigate this
     # problem, we will remove any reference to backports repos older than oldstable
   
     # https://github.com/GoogleCloudDataproc/initialization-actions/issues/1157
     oldstable=$(curl -s https://deb.debian.org/debian/dists/oldstable/Release | awk '/^Codename/ {print $2}');
     stable=$(curl -s https://deb.debian.org/debian/dists/stable/Release | awk '/^Codename/ {print $2}');
   
     matched_files="$(grep -rsil '\-backports' /etc/apt/sources.list*)"
     if [[ -n "$matched_files" ]]; then
       for filename in "$matched_files"; do
         grep -e "$oldstable-backports" -e "$stable-backports" "$filename" || \
           sed -i -e 's/^.*-backports.*$//' "$filename"
       done
     fi
   }
   
   function main() {
     OS=$(. /etc/os-release && echo "${ID}")
     if [[ ${OS} == debian ]] && [[ $(echo "${DATAPROC_IMAGE_VERSION} <= 2.1" | bc -l) == 1 ]]; then
       remove_old_backports
     fi
     recv_keys || err 'Unable to receive keys.'
     update_apt_get || err 'Unable to update packages lists.'
     install_kafka_python_package
   
     # Only run the installation on workers; verify zookeeper on master(s).
     if [[ "${ROLE}" == 'Master' ]]; then
       service zookeeper-server status ||
         err 'Required zookeeper-server not running on master!'
       if [[ "${RUN_ON_MASTER}" == "true" ]]; then
         # Run installation on masters.
         install_and_configure_kafka_server
       else
         # On master nodes, just install kafka command-line tools and libs but not
         # kafka-server.
         install_apt_get kafka ||
           err 'Unable to install kafka libraries on master!'
       fi
     else
       # Run installation on workers.
       install_and_configure_kafka_server
     fi
   }
   
   main
   

   
   

2. kafka.sh kopieren Initialisierungsaktion in Ihren Cloud Storage-Bucket ein. Dieses Skript installiert Kafka auf einem Dataproc-Cluster.

   1. Öffnen Sie Cloud Shell und führen Sie den folgenden Befehl aus: folgenden Befehl:

      gcloud storage cp gs://goog-dataproc-initialization-actions-REGION/kafka/kafka.sh gs://BUCKET_NAME/scripts/
      

      Ersetzen Sie die folgenden Werte:

      • REGION: kafka.sh ist öffentlich und regional mit Tags gespeichert Buckets in Cloud Storage. Standort in der Nähe angeben Compute Engine-Region (Beispiel: us-central1).
      • BUCKET_NAME: Der Name Ihres Cloud Storage-Buckets

Hinweis: Das Skript kafka.sh ist auf GitHub verfügbar. initialization-actions/kafka/ zu erstellen. Sie können das Skript klonen oder auf Ihren lokalen Computer herunterladen. Ändern Sie die im vorherigen gcloud storage cp-Beispiel angegebene Quelle zum Kopieren das lokale Skript zu gs://BUCKET_NAME/scripts/.

Dataproc-Kafka-Cluster erstellen

1. Öffnen Sie Cloud Shell und führen Sie den folgenden Befehl aus: die folgenden gcloud dataproc clusters create Befehl zum Erstellen einer Dataproc-Instanz Hochverfügbarkeitscluster Cluster, 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 enthalten Buchstaben, Ziffern und Bindestriche. Er darf nicht mit einem Bindestrich enden. Der Name eines kann ein gelöschter Cluster wiederverwendet werden.
   • PROJECT_ID: Das Projekt, das mit diesem Cluster verknüpft werden soll.
   • REGION: Die Compute Engine-Region in dem sich der Cluster befinden wird, z. B. us-central1.
     • Sie können das optionale Flag --zone=ZONE hinzufügen, zum Angeben einer Zone innerhalb der angegebenen Region, z. B. us-central1-a. Wenn Sie keine Zone angeben, Automatische Zonenplatzierung in Dataproc die Funktion eine Zone mit der angegebenen Region auswählt.
   • --image-version: Dataproc-Image-Version 2.1-debian11 wird für diese Anleitung empfohlen. Hinweis: Jede Image-Version enthält eine Reihe vorinstallierter Komponenten, einschließlich der Hive-Komponente, die in diesem (siehe Unterstützte Dataproc-Image-Versionen).
   • --num-master: 3 Master-Knoten erstellen ein Hochverfügbarkeitscluster. Die für Kafka erforderliche Zookeeper-Komponente ist vorinstalliert in einem Cluster mit Hochverfügbarkeit.
     <strong<Einstellung von Zookeeper<="" strong="">Künftig Kafka Releases, Zookeeper wird möglicherweise eingestellt und entfernt.</strong
   • --enable-component-gateway: Aktiviert die Dataproc-Komponentengateway.
   • BUCKET_NAME: der Name Ihres Cloud Storage-Bucket die das Initialisierungsskript /scripts/kafka.sh enthält Weitere Informationen finden Sie unter Kafka-Installationsskript in Cloud Storage kopieren.

Kafka-custdata-Thema erstellen

So erstellen Sie ein Kafka-Thema im Dataproc-Kafka-Cluster:

1. Verwenden Sie die SSH-Verbindung um ein Terminalfenster auf der Cluster-Master-VM zu öffnen.

2. 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, auf dem ausgeführt wird. Port 9092 auf dem worker-0-Knoten.

   • 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

Im folgenden Skript wird das Kafka-Tool kafka-console-producer.sh verwendet, um fiktive Kundendaten im CSV-Format generieren.

1. Kopieren Sie das Skript und fügen Sie es in den SSH-Tunnel ein. auf dem Masterknoten Ihres Kafka-Clusters. Drücken Sie &lt;return&gt;, 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.
   Sie können Daten zu einem Kafka-Thema aus vielen Quellen veröffentlichen. wie Websites, Anwendungen und Geräte, in verschiedenen Formaten wie CSV-, JSON-, Text- und Apache Avro-Serialisierungsformat (siehe Kafka-Anwendungsfälle)

2. Führen Sie den folgenden Kafka-Befehl aus, um zu prüfen, ob das Thema custdata enthält 10.000 Nachrichten.

   /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 ein cust_orc (ORC) Hive-Tabellen im Cloud Storage-Bucket.

1. Fügen Sie Ihren BUCKET_NAME in das folgende Skript ein: Kopieren Sie das Skript und fügen Sie es in das SSH-Terminal Ihres Kafka-Cluster-Masterknotens ein. Drücken Sie dann die &lt;return&gt;, um ein ~/hivetables.hql-Skript (Hive Query Language) zu erstellen.

   Das Skript ~/hivetables.hql wird ausgeführt. zur Erstellung von Parquet- und ORC Hive-Tabellen. in Ihrem Cloud Storage-Bucket.

   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
   

2. Führen Sie im SSH-Terminal auf dem Masterknoten Ihres Kafka-Clusters Hive-Job ~/hivetables.hql senden, um cust_parquet (parquet) und eine cust_orc (ORC) Hive-Tabelle in Ihrem Cloud Storage-Bucket.

   gcloud dataproc jobs submit hive \
       --cluster=KAFKA_CLUSTER \
       --region=REGION \
       -f ~/hivetables.hql
   

   Hinweise:

   • Die Hive-Komponente ist in Dataproc Kafka vorinstalliert Cluster. Siehe Releaseversionen 2.1.x finden Sie eine Liste der Versionen der Hive-Komponenten in kürzlich veröffentlichten 2.1-Images.
   • KAFKA_CLUSTER: Der Name Ihres Kafka-Clusters.
   • REGION: Die Region, in der sich Ihr Kafka-Cluster befindet.

Kafka-custdata in Hive-Tabellen streamen

1. Führen Sie im SSH-Terminal auf dem Masterknoten von Ihren Kafka-Cluster, um die kafka-python-Bibliothek zu installieren. Zum Streamen von Kafka-Themendaten ist ein Kafka-Client erforderlich Cloud Storage
   
   Normalerweise wird der Kafka-Client auf einem separaten Maschine, aber in dieser Anleitung wird ein Cluster für alle Prozesse vereinfachen und Kosten sparen.

   pip install kafka-python
   

2. Fügen Sie den folgenden Code (BUCKET_NAME) ein, kopieren Sie ihn und fügen Sie ihn dann ein Fügen Sie PySpark-Code in das SSH-Terminal Ihres Kafka-Cluster-Masterknotens ein. Drücken Sie dann die &lt;return&gt;, um eine streamdata.py-Datei zu erstellen.

   Das Skript abonniert das Kafka-Thema custdata und streamt dann Daten in Ihre Hive-Tabellen in Cloud Storage importieren. Das Ausgabeformat, Dies kann Parquet oder ORC sein, wird in das Skript als einen Parameter.

   Hinweis: Die an Cloud Storage gestreamten Kafka-Themendaten befinden sich in komprimierten Binärformat erstellt.

   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
   

3. Im SSH-Terminal auf dem Master-Knoten von Ihrem Kafka-Cluster, führen Sie spark-submit aus, um Daten Ihre Hive-Tabellen in Cloud Storage.

   1. Geben Sie den Namen Ihres KAFKA_CLUSTER und die Ausgabe ein FORMAT, kopieren Sie dann den folgenden Code und fügen Sie ihn in das SSH- auf dem Masterknoten Ihres Kafka-Clusters und drücken Sie dann &lt;return&gt;. um den Code auszuführen und die Kafka-custdata-Daten im Parquet-Format an Ihr Hive-Tabellen in 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
          

      Hinweise:

      • KAFKA_CLUSTER: Geben Sie den Namen Ihres Kafka-Clusters ein.
      • FORMAT: Geben Sie entweder parquet oder orc als . Sie können den Befehl nacheinander ausführen, beiden Formate in die Hive-Tabellen übertragen: So wird z. B. beim ersten Aufruf Geben Sie parquet an, um das Kafka-custdata-Thema an die Hive zu streamen Parkettboden Geben Sie dann beim zweiten Aufruf das Format orc an, Stream custdata in die Hive-ORC-Tabelle.

4. Wenn die Standardausgabe im SSH-Terminal angehalten wird, was bedeutet, dass alle custdata wurden gestreamt, drücken Sie &lt;control-c&gt; im SSH-Terminal, um den Prozess zu beenden.

5. Listen Sie die Hive-Tabellen in Cloud Storage auf.

   gcloud storage ls gs://BUCKET_NAME/tables/* --recursive
   

   Hinweise:

   • BUCKET_NAME: Geben Sie den Namen des Cloud Storage-Buckets ein. Bucket, der Ihre Hive-Tabellen enthält (siehe Hive-Tabellen erstellen).

Streamingdaten abfragen

1. Im SSH-Terminal auf dem Master-Knoten von Kafka-Cluster, führen Sie den folgenden hive-Befehl aus: zum Zählen der gestreamten Kafka-custdata-Nachrichten in den Hive-Tabellen in Cloud Storage.

   hive -e "select count(1) from TABLE_NAME"
   

   Hinweise:

   • TABLE_NAME: Geben Sie entweder cust_parquet oder cust_orc als Hive-Tabellenname.

   Erwartetes Ausgabe-Snippet:

...
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

Achtung: Das Löschen von Projekten hat folgende Auswirkungen:

• Alle Inhalte des Projekts werden gelöscht. Wenn Sie für die Aufgaben in diesem Dokument ein bereits bestehendes Projekt verwendet haben und dieses löschen, werden auch alle anderen im Rahmen des Projekts erstellten Daten gelöscht.
• Benutzerdefinierte Projekt-IDs gehen verloren. Beim Erstellen dieses Projekts haben Sie möglicherweise eine benutzerdefinierte Projekt-ID erstellt, die Sie weiterhin verwenden möchten. Damit die URLs, die die Projekt-ID nutzen, z. B. eine appspot.com-URL, erhalten bleiben, sollten Sie ausgewählte Ressourcen innerhalb des Projekts löschen, statt das gesamte Projekt.

Wenn Sie mehrere Architekturen, Anleitungen und Kurzanleitungen durcharbeiten möchten, können Sie die Überschreitung von Projektkontingenten verhindern, indem Sie Projekte wiederverwenden.

Google Cloud-Projekt löschen:

gcloud projects delete PROJECT_ID

Ressourcen löschen

• Löschen Sie den Bucket:

  gcloud storage buckets delete BUCKET_NAME

  Wichtig: Der Bucket muss leer sein, bevor Sie ihn löschen können.
• Löschen Sie den Kafka-Cluster:

  gcloud dataproc clusters delete KAFKA_CLUSTER \
      --region=${REGION}