Kafka-Thema an Hive streamen

Anleitungsschritte

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

Kafka-Installationsskript in Cloud Storage kopieren

Das kafka.sh Initialisierungsaktionsskript installiert Kafka auf einem Dataproc-Cluster.

1. Sehen Sie sich den Code an.
   

   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 a random number generated less than 10000.
       # 10000 is choosen since the max broker ID allowed being set is 10000.
       BROKER_ID=$((RANDOM % 10000))
     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. Kopieren Sie das Script kafka.sh für die Initialisierungsaktion in Ihren Cloud Storage-Bucket. Dieses Script installiert Kafka auf einem Dataproc-Cluster.

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

      gcloud storage 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, regional getaggten Cloud Storage-Buckets gespeichert. Geben Sie eine Compute Engine-Region an, die sich geografisch in der Nähe befindet (z. B. us-central1).
      • BUCKET_NAME: Der Name Ihres Cloud Storage-Buckets

Dataproc-Kafka-Cluster erstellen

1. Öffnen Sie Cloud Shell und führen Sie den folgenden gcloud dataproc clusters create-Befehl aus, um einen Dataproc-Cluster für hohe Verfügbarkeit zu erstellen, auf dem die Kafka- und ZooKeeper-Komponenten installiert werden:

   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 darf 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 diesem Cluster zugeordnet werden soll.
   • REGION: Die Compute Engine-Region, in der sich der Cluster befindet, 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, wird mit der automatischen Zonenplatzierung von Dataproc eine Zone mit der angegebenen Region ausgewählt.
   • --image-version: Für dieses Tutorial wird die Dataproc-Image-Version 2.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 Zookeeper-Komponente, die für Kafka erforderlich ist, ist in einem HA-Cluster vorinstalliert.
   • --enable-component-gateway: Aktiviert das Dataproc Component Gateway.
   • BUCKET_NAME: Der Name Ihres Cloud Storage-Bucket, der das /scripts/kafka.sh-Initialisierungsskript enthält (siehe Kafka-Installationsskript in Cloud Storage kopieren).

Kafka-custdata-Thema erstellen

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

1. Verwenden Sie das SSH-Dienstprogramm, um ein Terminalfenster auf der Clustermaster-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 steht für den Kafka-Broker, der auf Port 9092 des Knotens worker-0 ausgeführt wird.

   • Nachdem Sie das custdata-Thema 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 Script werden mit dem kafka-console-producer.sh-Kafka-Tool fiktive Kundendaten im CSV-Format generiert.

1. Kopieren Sie das Script und fügen Sie es in das SSH-Terminal auf dem Masterknoten Ihres Kafka-Clusters ein. Drücken Sie die Taste <return>, um das Script 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.

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

Erstellen Sie Hive-Tabellen, um gestreamte Kafka-Themendaten zu empfangen. Führen Sie die folgenden Schritte aus, um cust_parquet- (Parquet-) und cust_orc- (ORC-)Hive-Tabellen in Ihrem Cloud Storage-Bucket zu erstellen.

1. Fügen Sie Ihre BUCKET_NAME in das folgende Script ein, kopieren Sie das Script und fügen Sie es in das SSH-Terminal auf dem Masterknoten Ihres Kafka-Clusters ein. Drücken Sie dann die Eingabetaste, um ein ~/hivetables.hql-Script (Hive Query Language) zu erstellen.

   Im nächsten Schritt führen Sie das ~/hivetables.hql-Script 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
   

2. Reichen Sie im SSH-Terminal auf dem Masterknoten Ihres Kafka-Clusters den ~/hivetables.hql-Hive-Job ein, um cust_parquet- (Parquet) und cust_orc- (ORC)-Hive-Tabellen in Ihrem Cloud Storage-Bucket zu erstellen.

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

   Hinweise:

   • Die Hive-Komponente ist auf dem Dataproc-Kafka-Cluster vorinstalliert. Eine Liste der Hive-Komponentenversionen, die in den kürzlich veröffentlichten 2.1-Images enthalten sind, finden Sie unter 2.1.x-Releaseversionen.
   • 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 den folgenden Befehl im SSH-Terminal auf dem Masterknoten Ihres Kafka-Clusters aus, um die kafka-python-Bibliothek zu installieren. Ein Kafka-Client ist erforderlich, um Kafka-Themendaten in Cloud Storage zu streamen.
   

   pip install kafka-python
   

2. Fügen Sie Ihre BUCKET_NAME ein, kopieren Sie den folgenden PySpark-Code und fügen Sie ihn in das SSH-Terminal auf dem Masterknoten Ihres Kafka-Clusters ein. Drücken Sie dann die Eingabetaste, um eine streamdata.py-Datei zu erstellen.

   Das Script abonniert das Kafka-custdata-Thema und streamt die Daten dann in Ihre Hive-Tabellen in Cloud Storage. Das Ausgabeformat, das Parquet oder ORC sein kann, wird als Parameter an das Script ü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
   

3. Führen Sie im SSH-Terminal auf dem Masterknoten Ihres Kafka-Clusters spark-submit aus, um Daten an Ihre Hive-Tabellen in Cloud Storage zu streamen.

   1. Geben Sie den Namen Ihrer KAFKA_CLUSTER und die AusgabeFORMAT 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 die Eingabetaste, 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 oder orc als Ausgabeformat an. Sie können den Befehl nacheinander ausführen, um beide Formate in die Hive-Tabellen zu streamen. Geben Sie beispielsweise bei der ersten Ausführung parquet an, um das Kafka-Thema custdata in die Hive-Parquet-Tabelle zu streamen. Geben Sie dann bei der zweiten Ausführung das Format orc an, um custdata in die Hive-ORC-Tabelle zu streamen.

4. Wenn die Standardausgabe im SSH-Terminal beendet wird, was bedeutet, dass die gesamte custdata gestreamt wurde, drücken Sie im SSH-Terminal <Strg-C>, um den Vorgang 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, der Ihre Hive-Tabellen enthält (siehe Hive-Tabellen erstellen).

Gestreamte Daten abfragen

1. 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 oder cust_orc als Namen der Hive-Tabelle an.

   Beispiel für die erwartete Ausgabe:

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