Kafka-Thema an Hive streamen

Apache Kafka ist eine Open-Source-Plattform für verteiltes Streaming für Echtzeit-Datenpipelines und Datenintegration. Es bietet ein effizientes und skalierbares Streaming-System für eine Vielzahl von Anwendungen, darunter:

  • Echtzeitanalysen
  • Streamverarbeitung
  • Logaggregation
  • Verteilte Nachrichten
  • Event-Streaming

Ziele

  1. Installieren Sie Kafka in einem HA-Cluster für Managed Service for Apache Spark mit ZooKeeper (in dieser Anleitung als „Managed Service for Apache Spark Kafka-Cluster“ bezeichnet).

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

  3. Erstellen Sie Hive-Parquet- und ORC-Tabellen in Cloud Storage, um gestreamte Kafka-Themendaten zu empfangen.

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

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

Neuen Nutzern von Google Cloud 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.

Hinweis

Falls Sie es noch nicht getan haben, erstellen Sie ein Google Cloud -Projekt.

  1. Melden Sie sich in Ihrem Google Cloud -Konto an. Wenn Sie mit Google Cloudnoch 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.

    Roles required to select or create a project

    • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
    • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

    Go to project selector

  3. Verify that billing is enabled for your Google Cloud project.

  4. Enable the Dataproc, Compute Engine, and Cloud Storage APIs.

    Roles required to enable APIs

    To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

    Enable the APIs

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

    Roles required to select or create a project

    • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
    • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

    Go to project selector

  6. Verify that billing is enabled for your Google Cloud project.

  7. Enable the Dataproc, Compute Engine, and Cloud Storage APIs.

    Roles required to enable APIs

    To enable APIs, you need the Service Usage Admin IAM role (roles/serviceusage.serviceUsageAdmin), which contains the serviceusage.services.enable permission. Learn how to grant roles.

    Enable the APIs

    8.
  8. Wechseln Sie in der Google Cloud Console unter „Cloud Storage“ zur Seite Buckets.

    Buckets aufrufen

  9. Klicken Sie auf Erstellen.
  10. Geben Sie auf der Seite Bucket erstellen die Bucket-Informationen ein. Klicken Sie auf Weiter, um mit dem nächsten Schritt fortzufahren.
    1. Führen Sie im Abschnitt Einstieg die folgenden Schritte aus:
      • Geben Sie einen global eindeutigen Namen ein, der den Anforderungen für Bucket-Namen entspricht.
      • So fügen Sie ein Bucket-Label hinzu: Maximieren Sie den Abschnitt Labels (), klicken Sie auf Label hinzufügen und geben Sie key und value für Ihr Label an.
    2. Gehen Sie im Bereich Speicherort für Daten auswählen so vor:
      1. Standorttyp auswählen.
      2. Wählen Sie im Drop-down-Menü Standorttyp einen Standort aus, an dem die Daten Ihres Buckets dauerhaft gespeichert werden.
        • Wenn Sie den Standorttyp Dual-Region auswählen, können Sie auch die Turboreplikation aktivieren, indem Sie das entsprechende Kästchen anklicken.
      3. Wenn Sie die Bucket-übergreifende Replikation einrichten möchten, wählen Sie Bucket-übergreifende Replikation über Storage Transfer Service hinzufügen aus und führen Sie die folgenden Schritte aus:

        Bucket-übergreifende Replikation einrichten

        1. Wählen Sie im Menü Bucket einen Bucket aus.

        2. Klicken Sie im Bereich Replikationseinstellungen auf Konfigurieren, um die Einstellungen für den Replikationsjob zu konfigurieren.

          Der Bereich Bucket-übergreifende Replikation konfigurieren wird angezeigt.

          • Wenn Sie die zu replizierenden Objekte nach dem Objektnamenspräfix filtern möchten, geben Sie ein Präfix ein, mit dem Sie Objekte ein- oder ausschließen möchten, und klicken Sie dann auf  Präfix hinzufügen.
          • Wenn Sie eine Speicherklasse für die replizierten Objekte festlegen möchten, wählen Sie im Menü Speicherklasse eine Speicherklasse aus. Wenn Sie diesen Schritt überspringen, wird für replizierte Objekte standardmäßig die Speicherklasse des Ziel-Buckets verwendet.
          • Klicken Sie auf Fertig.
    3. Gehen Sie im Bereich Speicherort für Daten auswählen so vor:
      1. Wählen Sie eine Standardspeicherklasse für den Bucket oder Autoclass für die automatische Speicherklassenverwaltung der Daten Ihres Buckets aus.
      2. Wenn Sie den hierarchischen Namespace aktivieren möchten, wählen Sie im Bereich Speicher für datenintensive Arbeitslasten optimieren die Option Hierarchischen Namespace für diesen Bucket aktivieren aus.
    4. Wählen Sie im Abschnitt Zugriff auf Objekte steuern aus, ob der Bucket Verhinderung des öffentlichen Zugriffs durchsetzt, und wählen Sie eine Zugriffssteuerungsmethode für die Objekte Ihres Buckets aus.
    5. Führen Sie im Bereich Auswählen, wie Objektdaten geschützt werden die folgenden Schritte aus:
      • Wählen Sie unter Datenschutz die gewünschten Optionen für Ihren Bucket aus.
        • Wenn Sie Vorläufiges Löschen aktivieren möchten, klicken Sie das Kästchen Richtlinie für vorläufiges Löschen (zur Datenwiederherstellung) an und geben Sie die Anzahl der Tage an, die Objekte nach dem Löschen beibehalten werden sollen.
        • Wenn Sie die Objektversionsverwaltung festlegen möchten, klicken Sie das Kästchen Objektversionsverwaltung (zur Datenwiederherstellung) an und geben Sie die maximale Anzahl von Versionen pro Objekt und die Anzahl der Tage an, nach denen die nicht aktuellen Versionen ablaufen.
        • Klicken Sie das Kästchen Aufbewahrung (für Compliance) an, um die Aufbewahrungsrichtlinie für Objekte und Buckets zu aktivieren, und gehen Sie dann so vor:
          • Klicken Sie auf das Kästchen Objektaufbewahrung aktivieren, um die Objektaufbewahrungssperre zu aktivieren.
          • Wenn Sie Bucket Lock aktivieren möchten, klicken Sie das Kästchen Bucket-Aufbewahrungsrichtlinie festlegen an und wählen Sie eine Zeiteinheit und eine Zeitdauer für die Aufbewahrungsdauer aus.
      • Um auszuwählen, wie Ihre Objektdaten verschlüsselt werden, maximieren Sie den Bereich Datenverschlüsselung () und wählen Sie eine Methode für die Datenverschlüsselung aus.
  11. Klicken Sie auf Erstellen.

Anleitungsschritte

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

Kafka-Installationsskript in Cloud Storage kopieren

Mit dem kafka.sh-Initialisierungsaktionsskript wird Kafka auf einem Managed Service for Apache Spark-Cluster installiert.

  1. Sehen Sie sich den Code an. 

    #!/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() {
  if [[ ${OS} == debian ]] && [[ $(echo "${DATAPROC_IMAGE_VERSION} >= 3.0" | bc -l) == 1 ]]; then
    retry_apt_command "apt-get update && apt-get install -y gnupg"
    export GNUPGHOME="$(mktemp -d)"
    trap 'rm -rf "${GNUPGHOME}"' EXIT
    gpg --keyserver keyserver.ubuntu.com --recv-keys B7B3B788A8D3785C
    mkdir -p /etc/apt/trusted.gpg.d
    gpg --export B7B3B788A8D3785C > /etc/apt/trusted.gpg.d/mysql-repo.gpg
  else
    retry_apt_command "apt-get install -y gnupg2 && \
      apt-key adv --keyserver keyserver.ubuntu.com --recv-keys B7B3B788A8D3785C"
  fi
}

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 kafka.sh-Skript für die Initialisierungsaktion in Ihren Cloud Storage-Bucket. Mit diesem Skript wird Kafka auf einem Managed Service for Apache Spark-Cluster installiert.

    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 gekennzeichneten Buckets in Cloud Storage gespeichert. Geben Sie eine geografisch nahe Compute Engine-Region an (z. B. us-central1).
      • BUCKET_NAME: Der Name Ihres Cloud Storage-Buckets

Managed Service for Apache Spark-Kafka-Cluster erstellen

  1. Öffnen Sie Cloud Shell und führen Sie dann den folgenden gcloud dataproc clusters create-Befehl aus, um einen HA-Cluster-Cluster für Managed Service for Apache Spark zu erstellen, in 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. Das letzte Zeichen darf kein Bindestrich sein. 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 Funktion Automatische Zonenzuweisung von Managed Service for Apache Spark eine Zone in der angegebenen Region ausgewählt.
    • --image-version: Für dieses Tutorial wird die Managed Service for Apache Spark-Image-Version 2.1-debian11 empfohlen. Hinweis: Jede Image-Version enthält eine Reihe vorinstallierter Komponenten, einschließlich der in diesem Tutorial verwendeten Hive-Komponente (siehe Unterstützte Image-Versionen für Managed Service for Apache Spark).
    • --num-master: 3 Masterknoten bilden einen HA-Cluster. Die Zookeeper-Komponente, die für Kafka erforderlich ist, ist in einem HA-Cluster vorinstalliert.
    • --enable-component-gateway: Aktiviert das Managed Service for Apache Spark-Komponentengateway.
    • BUCKET_NAME: Der Name Ihres Cloud Storage-Bucket, der das /scripts/kafka.sh-Initialisierungsskript enthält (siehe Kafka-Installationsskript in Cloud Storage kopieren).

Kafka-Thema custdata erstellen

So erstellen Sie ein Kafka-Thema im Managed Service for Apache Spark-Kafka-Cluster:

  1. Verwenden Sie das SSH-Tool, um ein Terminalfenster auf der Master-VM des Clusters zu öffnen.

  2. Erstellen Sie ein Kafka-Thema custdata.

    /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 dem Knoten worker-0 auf Port 9092 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

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

  1. Kopieren Sie das Skript und fügen Sie es dann im SSH-Terminal auf dem Masterknoten Ihres Kafka-Clusters ein. Drücken Sie die Eingabetaste, 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.

  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 Skript ein, kopieren Sie es 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-Skript (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. Senden Sie im SSH-Terminal auf dem Masterknoten Ihres Kafka-Clusters den ~/hivetables.hql-Hive-Job, 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 im Managed Service for Apache Spark 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. Zum Streamen von Kafka-Themendaten in Cloud Storage ist ein Kafka-Client erforderlich. 
    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 Skript abonniert das Kafka-Thema custdata 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 in Ihre Hive-Tabellen in Cloud Storage zu streamen.

    1. Geben Sie den Namen Ihres KAFKA_CLUSTER und der Ausgabe FORMAT ein, kopieren Sie 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 Ausgabeforamt an. Sie können den Befehl nacheinander ausführen, um beide Formate in die Hive-Tabellen zu streamen. Geben Sie beispielsweise beim ersten Aufruf parquet an, um das Kafka-Thema custdata in die Hive-Parquet-Tabelle zu streamen. Geben Sie dann beim zweiten Aufruf das Format orc an, um custdata in die Hive-ORC-Tabelle zu streamen.

  4. Wenn die Standardausgabe im SSH-Terminal angehalten wird, was bedeutet, dass alle custdata gestreamt wurden, drücken Sie im SSH-Terminal <control-c>, um den Vorgang zu beenden.

  5. Hive-Tabellen in Cloud Storage auflisten

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

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

    Erwarteter Auszug der 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)

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
  • Kafka-Cluster löschen: 
    gcloud dataproc clusters delete KAFKA_CLUSTER \
    --region=${REGION}