将 Kafka 主题流式传输到 Hive

Apache Kafka 是一个开源分布式流式传输平台，用于实时数据流水线和数据集成。它提供了一个高效且可扩缩的流式传输系统，可用于各种应用，包括：

  • 实时分析
  • 流处理
  • 日志汇总
  • 分布式消息传递
  • 事件流处理

目标

  1. 在带有 ZooKeeper 的 Managed Service for Apache Spark 高可用性集群（在本教程中称为“Managed Service for Apache Spark Kafka 集群”）上安装 Kafka。

  2. 创建虚构的客户数据，然后将该数据发布到 Kafka 主题。

  3. 在 Cloud Storage 中创建 Hive Parquet 和 ORC 表，以接收流式传输的 Kafka 主题数据。

  4. 提交 PySpark 作业，以订阅 Kafka 主题并以 Parquet 和 ORC 格式将该主题流式传输到 Cloud Storage。

  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 serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). 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 serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

    Enable the APIs

    8.
  8. 在 Google Cloud 控制台中，前往 Cloud Storage 存储分区页面。

    进入“存储桶”

  9. 点击 创建。
  10. 在创建存储桶页面上，输入您的存储桶信息。要转到下一步，请点击继续。
    1. 在开始使用部分中，执行以下操作：
      • 输入符合存储桶命名要求的全局唯一的名称。
      • 如需添加存储桶标签，请展开标签部分 ()，点击 添加标签，并为标签指定 key 和 value。
    2. 在选择数据存储位置部分，执行以下操作：
      1. 选择位置类型。
      2. 从位置类型下拉菜单中选择一个位置，用于永久存储存储桶的数据。
      3. 如需设置跨存储桶复制，请选择通过 Storage Transfer Service 添加跨存储桶复制，然后按照以下步骤操作：

        设置跨存储桶复制

        1. 在存储桶菜单中，选择一个存储桶。

        2. 在复制设置部分中，点击配置以配置复制作业的设置。

          系统会显示配置跨存储桶复制窗格。

          • 如需按对象名称前缀过滤要复制的对象，请输入要用于包含或排除对象的前缀，然后点击 添加前缀。
          • 如需为复制的对象设置存储类别，请从存储类别菜单中选择一个存储类别。 如果您跳过此步骤，则复制的对象会默认使用目标存储桶的存储类别。
          • 点击完成。
    3. 在选择数据存储方式部分中，执行以下操作：
      1. 为存储桶选择默认存储类别，或者选择 Autoclass 对存储桶数据进行自动存储类别管理。
      2. 如需启用分层命名空间，请在针对数据密集型工作负载优化存储部分中，选择在此存储桶上启用分层命名空间。
    4. 在选择如何控制对对象的访问权限部分中，选择存储桶是否强制执行禁止公开访问，然后为存储桶对象选择访问权限控制方法。
    5. 在选择如何保护对象数据部分中，执行以下操作：
      • 在数据保护下，选择您要为存储桶设置的任何选项。
        • 如需启用软删除，请选中软删除政策（用于数据恢复）复选框，然后指定您希望在删除对象后保留对象的天数。
        • 如需设置对象版本控制，请选中对象版本控制（用于版本控制）复选框，然后指定每个对象的最大版本数以及非当前版本过期前的天数。
        • 如需为对象和存储分区启用保留政策，请点击保留（合规性）复选框，然后执行以下操作：
          • 如需启用对象保留锁定，请点击启用对象保留复选框。
          • 如需启用存储桶锁定，请点击设置存储桶保留政策复选框，然后为保留期限选择时间单位和时长。
      • 如需选择对象数据的加密方式，请展开数据加密部分 ()，然后选择数据加密方法。
  11. 点击创建。

教程步骤

请执行以下步骤创建 Managed Service for Apache Spark Kafka 集群，以 Parquet 或 ORC 格式将 Kafka 主题读入 Cloud Storage。

将 Kafka 安装脚本复制到 Cloud Storage

kafka.sh 初始化操作脚本会在 Managed Service for Apache Spark 集群上安装 Kafka。

  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 存储桶中。此脚本会在 Managed Service for Apache Spark 集群上安装 Kafka。

    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 存储桶的名称。

创建 Managed Service for Apache Spark Kafka 集群

  1. 打开 Cloud Shell，然后运行以下 gcloud dataproc clusters create 命令，以创建用于安装 Kafka 和 ZooKeeper 组件的 Managed Service for Apache Spark 高可用性集群：

    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 主节点用于创建高可用性集群。Kafka 所需的 Zookeeper 组件已预安装在高可用性集群上。
    • --enable-component-gateway：启用 Managed Service for Apache Spark 组件网关。
    • BUCKET_NAME：包含 /scripts/kafka.sh 初始化脚本的 Cloud Storage 存储桶的名称（请参阅将 Kafka 安装脚本复制到 Cloud Storage）。

创建 Kafka custdata 主题

如需在 Managed Service for Apache Spark Kafka 集群上创建 Kafka 主题，请执行以下操作：

  1. 使用 SSH 实用程序在集群主虚拟机上打开终端窗口。

  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 表示在 worker-0 节点的端口 9092 上运行的 Kafka 代理。

    • 创建 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-console-producer.sh Kafka 工具生成 CSV 格式的虚构客户数据。

  1. 将该脚本复制并其粘贴到 Kafka 集群主节点上的 SSH 终端。按 <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

在 Cloud Storage 中创建 Hive 表

创建 Hive 表以接收流式传输的 Kafka 主题数据。 执行以下步骤，以在 Cloud Storage 存储桶中创建 cust_parquet (Parquet) 和 cust_orc (ORC) Hive 表。

  1. 在以下脚本中插入 BUCKET_NAME，将该脚本复制并粘贴到 Kafka 集群主节点上的 SSH 终端，然后按 <return> 以创建 ~/hivetables.hql（Hive 查询语言）脚本。

    您将在下一步中运行 ~/hivetables.hql 脚本，以在 Cloud Storage 存储桶中创建 Parquet 和 ORC Hive 表。

    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. 在 Kafka 集群主节点上的 SSH 终端中，提交 ~/hivetables.hql Hive 作业，以在 Cloud Storage 存储桶中创建 cust_parquet (Parquet) 和 cust_orc (ORC) Hive 表。

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

    注意：

    • Hive 组件已预安装在 Managed Service for Apache Spark Kafka 集群上。如需查看最近发布的 2.1 映像中包含的 Hive 组件版本列表，请参阅 2.1.x 发布版本。
    • KAFKA_CLUSTER：Kafka 集群的名称。
    • REGION：Kafka 集群所在的区域。

将 Kafka custdata 流式传输到 Hive 表

  1. 在 Kafka 集群的主节点上的 SSH 终端中运行以下命令，以安装 kafka-python 库。需要使用 Kafka 客户端才能将 Kafka 主题数据流式传输到 Cloud Storage。 
    pip install kafka-python

  2. 插入 BUCKET_NAME，将以下 PySpark 代码复制并粘贴到 Kafka 集群主节点上的 SSH 终端，然后按 <return> 以创建 streamdata.py 文件。

    该脚本会订阅 Kafka custdata 主题，然后将数据流式传输到 Cloud Storage 中的 Hive 表。输出格式（可以是 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. 在 Kafka 集群主节点上的 SSH 终端中，运行 spark-submit 以将数据流式传输到 Cloud Storage 中的 Hive 表。

    1. 插入 KAFKA_CLUSTER 的名称和输出 FORMAT，将以下代码复制并粘贴到 Kafka 集群主节点上的 SSH 终端，然后按 <return> 运行代码，并将 Kafka custdata 数据以 Parquet 格式流式传输到 Cloud Storage 中的 Hive 表。 

      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 以将 Kafka custdata 主题流式传输到 Hive Parquet 表；然后在第二次调用中，指定 orc 格式以将 custdata 流式传输到 Hive ORC 表。

  4. 标准输出在 SSH 终端中停止后（这表示所有 custdata 都已流式传输），请在 SSH 终端中按 <control-c> 以停止流程。

  5. 列出 Cloud Storage 中的 Hive 表。

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

    注意：

    • BUCKET_NAME：插入包含 Hive 表的 Cloud Storage 存储桶的名称（请参阅创建 Hive 表）。

查询流式传输的数据

  1. 在 Kafka 集群主节点上的 SSH 终端中，运行以下 hive 命令，以统计 Cloud Storage 的 Hive 表中流式传输的 Kafka custdata 消息的数量。

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

    注意：

    • TABLE_NAME：将 cust_parquet 或 cust_orc 指定为 Hive 表名称。

    预期输出代码段：

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