הזרמת נושא Kafka ל-Hive

‫Apache Kafka היא פלטפורמת סטרימינג מבוזרת בקוד פתוח לצינורות עיבוד נתונים בזמן אמת ולשילוב נתונים. היא מספקת מערכת סטרימינג יעילה וניתנת להרחבה לשימוש במגוון אפליקציות, כולל:

  • דיווח בזמן אמת
  • עיבוד זרמי נתונים (stream processing)
  • צבירת יומנים
  • הודעות מופצות
  • סטרימינג של אירועים

מטרות

  1. מתקינים את Kafka באשכול HA של Managed Service for Apache Spark עם ZooKeeper (שנקרא במדריך הזה 'אשכול Kafka של Managed Service for Apache Spark').

  2. יוצרים נתוני לקוחות פיקטיביים ואז מפרסמים את הנתונים בנושא Kafka.

  3. יצירה של טבלאות Hive parquet ו-ORC ב-Cloud Storage כדי לקבל נתונים מנושא Kafka בסטרימינג.

  4. שליחת משימת PySpark להרשמה לנושא Kafka ולהזרמת הנתונים ל-Cloud Storage בפורמט parquet ו-ORC.

  5. מריצים שאילתה על נתוני הטבלה של Hive שהועברו בסטרימינג כדי לספור את ההודעות של Kafka שהועברו בסטרימינג.

עלויות

במסמך הזה משתמשים ברכיבים הבאים של Google Cloud, והשימוש בהם כרוך בתשלום:

כדי להעריך את ההוצאות בהתאם לתחזית השימוש שלכם, אתם יכולים להיעזר במחשבון העלויות.

משתמשים חדשים של Google Cloud ? יכול להיות שאתם זכאים לתקופת ניסיון בחינם.

כשמסיימים את המשימות שמתוארות במסמך הזה אפשר למחוק את המשאבים שיצרתם כדי להימנע מחיובים נוספים. מידע נוסף זמין בקטע הסרת המשאבים.

לפני שמתחילים

אם עדיין לא עשיתם זאת, יוצרים Google Cloud פרויקט.

  1. נכנסים לחשבון Google Cloud . אם אתם משתמשים חדשים ב- Google Cloud, צרו חשבון כדי שתוכלו להעריך את הביצועים של המוצרים שלנו בתרחישים מהעולם האמיתי. לקוחות חדשים מקבלים בחינם גם קרדיט בשווי 300$ להרצה, לבדיקה ולפריסה של עומסי העבודה.

  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. במסוף Google Cloud , נכנסים לדף Buckets של Cloud Storage.

    כניסה לדף Buckets

  9. לוחצים על יצירה.
  10. ממלאים את פרטי הקטגוריה בדף Create a bucket. כדי לעבור לשלב הבא לוחצים על Continue.
    1. בקטע Get started (תחילת העבודה), מבצעים את הפעולות הבאות:
    2. בקטע Choose where to store your data, מבצעים את הפעולות הבאות:
      1. בוחרים סוג מיקום.
      2. בתפריט הנפתח Location type, בוחרים מיקום שבו יישמרו נתוני הקטגוריה באופן קבוע.
      3. כדי להגדיר שכפול בין מאגרי מידע, בוחרים באפשרות הוספת שכפול בין מאגרי מידע באמצעות Storage Transfer Service ופועלים לפי השלבים הבאים:

        הגדרה של רפליקציה בין מאגרי מידע

        1. בתפריט Bucket, בוחרים באפשרות הרצויה.

        2. בקטע הגדרות השכפול, לוחצים על הגדרה כדי להגדיר את ההגדרות של משימת השכפול.

          מופיעה החלונית Configure cross-bucket replication.

          • כדי לסנן אובייקטים לשכפול לפי קידומת של שם האובייקט, מזינים קידומת שרוצים לכלול או להחריג אובייקטים ממנה, ואז לוחצים על הוספת קידומת.
          • כדי להגדיר סוג אחסון לאובייקטים המשוכפלים, בוחרים סוג אחסון בתפריט סוג אחסון. אם מדלגים על השלב הזה, האובייקטים המשוכפלים ישתמשו בסוג האחסון של קטגוריית היעד כברירת מחדל.
          • לוחצים על סיום.
    3. בקטע Choose how to store your data, מבצעים את הפעולות הבאות:
      1. בוחרים default storage class לקטגוריה או Autoclass לניהול אוטומטי של סוג האחסון (storage class) של נתוני הקטגוריה.
      2. כדי להפעיל מרחב שמות היררכי, בקטע Optimize storage for data-intensive workloads, בוחרים באפשרות Enable hierarchical namespace on this bucket.
    4. בקטע Choose how to control access to objects, בוחרים אם הקטגוריה אוכפת public access prevention או לא, ואז בוחרים שיטת בקרת גישה לאובייקטים של הקטגוריה.
    5. בקטע Choose how to protect object data, מבצעים את הפעולות הבאות:
      • בוחרים באחת מהאפשרויות בקטע הגנה על נתונים שרוצים להגדיר לקטגוריה.
        • כדי להפעיל מחיקה עם יכולת שחזור, מסמנים את התיבה מדיניות מחיקה עם יכולת שחזור (לשחזור נתונים) ומציינים את מספר הימים שבהם רוצים לשמור אובייקטים אחרי המחיקה.
        • כדי להגדיר ניהול גרסאות של אובייקטים, מסמנים את התיבה ניהול גרסאות של אובייקטים (לשליטה בגרסאות) ומציינים את מספר הגרסאות המקסימלי לכל אובייקט ואת מספר הימים שאחריהם הגרסאות הלא עדכניות יפוגו.
        • כדי להפעיל את מדיניות שמירת הנתונים על אובייקטים וקטגוריות, לוחצים על תיבת הסימון שמירת נתונים (לצורך תאימות), ואז מבצעים את הפעולות הבאות:
          • כדי להפעיל את הנעילה של שמירת אובייקטים, מסמנים את התיבה הפעלת שמירת אובייקטים.
          • כדי להפעיל את נעילת הקטגוריה, מסמנים את תיבת הסימון הגדרת מדיניות שמירת נתונים בקטגוריה ובוחרים יחידת זמן ואת משך הזמן של תקופת השמירה.
      • כדי לבחור איך להצפין את נתוני האובייקט, מרחיבים את הקטע Data encryption () ובוחרים Data encryption method.
  11. לוחצים על יצירה.

שלבים במדריך

כדי ליצור אשכול Kafka של Managed Service for Apache Spark כדי לקרוא נושא Kafka ל-Cloud Storage בפורמט Parquet או ORC, צריך לבצע את השלבים הבאים.

העתקת סקריפט ההתקנה של Kafka ל-Cloud Storage

הסקריפט של kafka.sh פעולת האתחול מתקין את Kafka באשכול Managed Service for Apache Spark.

  1. מעיינים בקוד. 

    #!/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. מעתיקים את הסקריפט של kafka.sh פעולת ההפעלה לקטגוריה של Cloud Storage. הסקריפט הזה מתקין את Kafka באשכול Managed Service for Apache Spark.

    1. פותחים את Cloud Shell ומריצים את הפקודה הבאה: 

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

      מחליפים את הפרטים הבאים:

      • ‫REGION: kafka.sh מאוחסן בדליים ציבוריים עם תיוג אזורי ב-Cloud Storage. מציינים אזור Compute Engine שקרוב מבחינה גיאוגרפית (לדוגמה: us-central1).
      • ‫BUCKET_NAME: שם הקטגוריה של Cloud Storage.

יצירת אשכול Kafka של Managed Service for Apache Spark

  1. פותחים את Cloud Shell ומריצים את הפקודה הבאה gcloud dataproc clusters create כדי ליצור אשכול HA של Managed Service for Apache Spark שמותקנים בו הרכיבים Kafka ו-ZooKeeper:

    gcloud dataproc clusters create KAFKA_CLUSTER \
    --project=PROJECT_ID \
    --region=REGION \
    --image-version=2.1-debian11 \
    --num-masters=3 \
    --enable-component-gateway \
    --initialization-actions=gs://BUCKET_NAME/scripts/kafka.sh

    הערות:

    • ‫KAFKA_CLUSTER: שם האשכול, שחייב להיות ייחודי בתוך פרויקט. השם חייב להתחיל באות קטנה, ויכול להכיל עד 51 אותיות קטנות, מספרים ומקפים. הוא לא יכול להסתיים במקף. אפשר לעשות שימוש חוזר בשם של אשכול שנמחק.
    • ‫PROJECT_ID: הפרויקט שאליו רוצים לשייך את האשכול.
    • ‫REGION: האזור של Compute Engine שבו ימוקם האשכול, למשל us-central1.
      • אפשר גם להוסיף את הדגל האופציונלי --zone=ZONE כדי לציין אזור בתוך האזור שצוין, כמו us-central1-a. אם לא מציינים אזור, התכונה מיקום אוטומטי באזור של Managed Service for Apache Spark בוחרת אזור עם האזור שצוין.
    • --image-version: מומלץ להשתמש במדריך הזה בגרסת תמונה של Managed Service for Apache Spark 2.1-debian11. הערה: כל גרסת תמונה מכילה קבוצה של רכיבים שהותקנו מראש, כולל רכיב Hive שמשמש במדריך הזה (ראו גרסאות תמונה נתמכות של Managed Service for Apache Spark).
    • ‫--num-master: 3 צמתי בקרה יוצרים אשכול HA. רכיב Zookeeper, שנדרש על ידי Kafka, מותקן מראש באשכול HA. ‫
    • ‫--enable-component-gateway: הפעלה של Managed Service for Apache Spark Component Gateway.
    • ‫BUCKET_NAME: שם הקטגוריה של Cloud Storage שמכילה את /scripts/kafka.sh סקריפט ההפעלה/scripts/kafka.sh (ראו העתקת סקריפט ההתקנה של Kafka אל Cloud Storage).

יצירת נושא Kafka‏ custdata

כדי ליצור נושא Kafka באשכול Kafka של Managed Service for Apache Spark:

  1. משתמשים בכלי SSH כדי לפתוח חלון טרמינל במכונה הווירטואלית של מאסטר אשכולות (cluster master).

  2. יוצרים נושא Kafka custdata.

    /usr/lib/kafka/bin/kafka-topics.sh \
    --bootstrap-server KAFKA_CLUSTER-w-0:9092 \
    --create --topic custdata

    הערות:

    • ‫KAFKA_CLUSTER: מזינים את השם של אשכול Kafka. ‫-w-0:9092 מציין את ברוקר Kafka שפועל ביציאה 9092 בצומת worker-0.

    • אחרי שיוצרים את הנושא custdata, אפשר להריץ את הפקודות הבאות:

      # List all topics.
/usr/lib/kafka/bin/kafka-topics.sh \
    --bootstrap-server KAFKA_CLUSTER-w-0:9092 \
    --list

      
# Consume then display topic data.
/usr/lib/kafka/bin/kafka-console-consumer.sh \
    --bootstrap-server KAFKA_CLUSTER-w-0:9092 \
    --topic custdata

      
# Count the number of messages in the topic.
/usr/lib/kafka/bin/kafka-run-class.sh kafka.tools.GetOffsetShell \
    --broker-list KAFKA_CLUSTER-w-0:9092 \
    --topic custdata

      
# Delete topic.
/usr/lib/kafka/bin/kafka-topics.sh \
    --bootstrap-server KAFKA_CLUSTER-w-0:9092 \
    --delete --topic custdata

פרסום תוכן בנושא Kafka custdata

בסקריפט הבא נעשה שימוש בכלי Kafka כדי ליצור נתוני לקוחות פיקטיביים בפורמט CSV.kafka-console-producer.sh

  1. מעתיקים את הסקריפט ומדביקים אותו במסוף ה-SSH בצומת הראשי של אשכול Kafka. לוחצים על <return> כדי להריץ את הסקריפט.

    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=:"

    הערות:

    • ‫KAFKA_CLUSTER: השם של אשכול Kafka.

  2. מריצים את פקודת Kafka הבאה כדי לוודא שהנושא custdata מכיל 10,000 הודעות.

    /usr/lib/kafka/bin/kafka-run-class.sh kafka.tools.GetOffsetShell \
--broker-list KAFKA_CLUSTER-w-0:9092 \
--topic custdata

    הערות:

    • ‫KAFKA_CLUSTER: השם של אשכול Kafka.

    הפלט אמור להיראות כך: 

    custdata:0:10000

יצירת טבלאות Hive ב-Cloud Storage

יוצרים טבלאות Hive כדי לקבל נתונים מנושא Kafka שמוזרמים. כדי ליצור טבלאות Hive בפורמט cust_parquet (parquet) ו-cust_orc (ORC) בקטגוריה של Cloud Storage, מבצעים את השלבים הבאים.

  1. מזינים את BUCKET_NAME בסקריפט הבא, מעתיקים ומדביקים את הסקריפט בטרמינל SSH בצומת הראשי של אשכול Kafka, ואז לוחצים על <return> כדי ליצור סקריפט ~/hivetables.hql (Hive Query Language).

    בשלב הבא תריצו את סקריפט ~/hivetables.hql כדי ליצור טבלאות Hive בפורמטים Parquet ו-ORC בקטגוריה שלכם ב-Cloud Storage.

    cat > ~/hivetables.hql <<EOF
drop table if exists cust_parquet;
create external table if not exists cust_parquet
(uuid string, custname string, age string, amount string)
row format delimited fields terminated by ','
stored as parquet
location "gs://BUCKET_NAME/tables/cust_parquet";

    
drop table if exists cust_orc;
create external table if not exists cust_orc
(uuid string, custname string, age string, amount string)
row format delimited fields terminated by ','
stored as orc
location "gs://BUCKET_NAME/tables/cust_orc";
EOF

  2. בטרמינל SSH בצומת הראשי של אשכול Kafka, שולחים את משימת Hive‏ ~/hivetables.hql כדי ליצור טבלאות Hive‏ cust_parquet (parquet) ו-cust_orc (ORC) בקטגוריה של Cloud Storage.

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

    הערות:

    • רכיב Hive מותקן מראש באשכול Managed Service for Apache Spark Kafka. לרשימה של גרסאות רכיבי Hive שכלולות בתמונות 2.1 שפורסמו לאחרונה, אפשר לעיין בגרסאות 2.1.x.
    • ‫KAFKA_CLUSTER: השם של אשכול Kafka.
    • ‫REGION: האזור שבו נמצא אשכול Kafka.

הזרמת נתונים מ-Kafka‏ custdata לטבלאות Hive

  1. כדי להתקין את ספריית kafka-python, מריצים את הפקודה הבאה בטרמינל SSH בצומת הראשי של אשכול Kafka: כדי להזרים נתונים מנושא Kafka ל-Cloud Storage, צריך לקוח Kafka. 
    pip install kafka-python

  2. מזינים את BUCKET_NAME, מעתיקים ומדביקים את קוד PySpark הבא בטרמינל SSH בצומת הראשי של מאסטר אשכולות (cluster master) Kafka, ואז לוחצים על <return> כדי ליצור קובץ streamdata.py.

    הסקריפט נרשם לנושא custdata ב-Kafka, ואז מעביר את הנתונים בסטרימינג לטבלאות Hive ב-Cloud Storage. פורמט הפלט, שיכול להיות parquet או ORC, מועבר לסקריפט כפרמטר.

    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. במסוף SSH בצומת הראשי של אשכול Kafka, מריצים את הפקודה spark-submit כדי להזרים נתונים לטבלאות Hive ב-Cloud Storage.

    1. מזינים את השם של KAFKA_CLUSTER ואת הפלט FORMAT, מעתיקים ומדביקים את הקוד הבא במסוף ה-SSH בצומת הראשי של אשכול Kafka, ואז לוחצים על <return> כדי להריץ את הקוד ולשדר את נתוני Kafka custdata בפורמט parquet לטבלאות Hive ב-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

      הערות:

      • ‫KAFKA_CLUSTER: מזינים את השם של אשכול Kafka.
      • ‫FORMAT: מציינים את פורמט הפלט כ-parquet או כ-orc. אפשר להריץ את הפקודה ברצף כדי להזרים את שני הפורמטים לטבלאות Hive: לדוגמה, בהפעלה הראשונה, מציינים parquet כדי להזרים את הנושא custdata של Kafka לטבלת parquet של Hive; לאחר מכן, בהפעלה השנייה, מציינים את הפורמט orc כדי להזרים את custdata לטבלת ORC של Hive.

  4. אחרי שהפלט הרגיל נעצר בטרמינל של SSH, מה שמציין שכל הנתונים של custdata הועברו בסטרימינג, מקישים על Control-C בטרמינל של SSH כדי לעצור את התהליך.

  5. הצגת רשימה של טבלאות Hive ב-Cloud Storage.

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

    הערות:

    • ‫BUCKET_NAME: מזינים את השם של קטגוריית Cloud Storage שמכילה את טבלאות Hive (ראו יצירת טבלאות Hive).

שאילתות על נתונים שמוזרמים

  1. בטרמינל של SSH בצומת הראשי של אשכול Kafka, מריצים את הפקודה hive הבאה כדי לספור את ההודעות של Kafka custdata שמוזרמות בטבלאות Hive ב-Cloud Storage.

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

    הערות:

    • ‫TABLE_NAME: מציינים את שם טבלת Hive, למשל cust_parquet או cust_orc.

    קטע מהפלט הצפוי:

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

הסרת המשאבים

מחיקת הפרויקט

    כדי למחוק Google Cloud פרויקט:

    gcloud projects delete PROJECT_ID

מחיקת משאבים

  • מוחקים את הקטגוריה: 
    gcloud storage buckets delete BUCKET_NAME
  • מחיקת אשכול Kafka: 
    gcloud dataproc clusters delete KAFKA_CLUSTER \
    --region=${REGION}