Apache Kafka 是一个开源分布式流式处理平台，适用于实时数据流水线和数据集成。它提供了一个高效且可伸缩的流处理系统，用于各种应用，包括：
- 实时分析
- 流处理
- 日志汇总
- 分布式消息传递
- 事件流式传输
目标
使用 ZooKeeper（本教程中称为“Dataproc Kafka 集群”）在 Dataproc 高可用性集群上安装 Kafka。
创建虚构的客户数据，然后将数据发布到 Kafka 主题。
在 Cloud Storage 中创建 Hive Parquet 表和 ORC 表，以接收流式 Kafka 主题数据。
提交 PySpark 作业以订阅 Kafka 主题，并以 Parquet 和 ORC 格式将 Kafka 主题流式传输到 Cloud Storage。
对流式传输的 Hive 表数据运行查询，以统计流式传输的 Kafka 消息。
费用
在本文档中，您将使用 Google Cloud 的以下收费组件：
您可使用价格计算器根据您的预计使用情况来估算费用。
完成本文档中描述的任务后，您可以通过删除所创建的资源来避免继续计费。如需了解详情，请参阅清理。
准备工作
如果您尚未创建 Google Cloud 项目，请先创建一个。
- 登录您的 Google Cloud 账号。如果您是 Google Cloud 新手，请创建一个账号来评估我们的产品在实际场景中的表现。新客户还可获享 $300 赠金，用于运行、测试和部署工作负载。
-
In the Google Cloud console, on the project selector page, select or create a Google Cloud project.
-
启用 Dataproc, Compute Engine, and Cloud Storage API。
-
In the Google Cloud console, on the project selector page, select or create a Google Cloud project.
-
启用 Dataproc, Compute Engine, and Cloud Storage API。
- 在 Google Cloud 控制台中，进入 Cloud Storage 存储桶页面。
- 点击创建存储分区。
- 在创建存储分区页面上，输入您的存储分区信息。要转到下一步，请点击继续。
- 点击创建。
教程步骤
执行以下步骤以创建 Dataproc Kafka 集群，从而以 Parquet OR ORC 格式将 Kafka 主题读取到 Cloud Storage 中。
将 Kafka 安装脚本复制到 Cloud Storage
kafka.sh 初始化操作脚本会在 Dataproc 集群上安装 Kafka。
浏览代码。
将
kafka.sh初始化操作脚本复制到您的 Cloud Storage 存储桶。此脚本在 Dataproc 集群上安装 Kafka。打开 Cloud Shell，然后运行以下命令：
gsutil 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 存储桶的名称。
- REGION：
创建 Dataproc Kafka 集群
打开 Cloud Shell，然后运行以下
gcloud dataproc clusters create命令以创建会安装 Kafka 和 ZooKeeper 组件的 Dataproc 高可用性集群集群：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。如果您未指定可用区，Dataproc 自动选择可用区功能将选择具有指定区域的可用区。
- 您可以添加可选的
--image-version：本教程推荐使用 Dataproc 映像版本2.1-debian11。注意：每个映像版本都包含一组预安装的组件，包括本教程中使用的 Hive 组件（请参阅支持的 Dataproc 映像版本）。--num-master：3个主节点会创建一个高可用性集群。Kafka 所需的 ZooKeeper 组件已预安装在高可用性集群中。--enable-component-gateway：启用 Dataproc 组件网关。- BUCKET_NAME：包含
/scripts/kafka.sh初始化脚本的 Cloud Storage 存储桶的名称（请参阅将 Kafka 安装脚本复制到 Cloud Storage）。
创建 Kafka custdata 主题
如需在 Dataproc Kafka 集群上创建 Kafka 主题，请执行以下操作：
使用 SSH 实用程序在集群主服务器虚拟机上打开一个终端窗口。
创建 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 Broker。您可以在创建
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 格式的虚构客户数据。
复制该脚本，然后将其粘贴到 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 集群的名称。
运行以下 Kafka 命令以确认
custdata主题包含 10000 条消息。/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 表。
在以下脚本中插入 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
在 Kafka 集群主节点的 SSH 终端上，提交
~/hivetables.hqlHive 作业，以在 Cloud Storage 存储桶中创建cust_parquet(parquet) 和cust_orc(ORC) Hive 表。gcloud dataproc jobs submit hive \ --cluster=KAFKA_CLUSTER \ --region=REGION \ -f ~/hivetables.hql
备注：
- Hive 组件已预安装在 Dataproc Kafka 集群上。如需查看最近发布的 2.1 映像中包含的 Hive 组件版本列表，请参阅 2.1.x 发布版本。
- KAFKA_CLUSTER：您的 Kafka 集群的名称。
- REGION：您的 Kafka 集群所在的区域。
将 Kafka custdata 流式传输到 Hive 表
- 在 Kafka 集群主节点上的 SSH 终端中运行以下命令，以安装
kafka-python库。需要 Kafka 客户端才能将 Kafka 主题数据流式传输到 Cloud Storage。
pip install kafka-python
插入 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在 Kafka 集群主节点上的 SSH 终端中，运行
spark-submit以将数据流式传输到 Cloud Storage 中的 Hive 表。插入 KAFKA_CLUSTER 的名称和输出 FORMAT，然后复制以下代码并将其粘贴到 Kafka 集群主节点的 SSH 终端，然后按 <return> 运行该代码，并以 Parquet 格式将 Kafka
custdata数据流式传输到 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以将 Kafkacustdata主题流式传输到 Hive Parquet 表；然后在第二次调用中，指定orc格式以将custdata流式传输到 Hive ORC 表。
当标准输出在 SSH 终端停止（这表示所有
custdata已流式传输）后，在 SSH 终端中按 <control-c> 以停止该进程。列出 Cloud Storage 中的 Hive 表。
gsutil ls -r gs://BUCKET_NAME/tables/*
备注：
- BUCKET_NAME：插入包含 Hive 表的 Cloud Storage 存储桶的名称（请参阅创建 Hive 表）。
查询流式数据
在 Kafka 集群主节点上的 SSH 终端中，运行以下
hive命令以统计 Cloud Storage 的 Hive 表中流式传输的 Kafkacustdata消息。hive -e "select count(1) from TABLE_NAME"
备注：
- TABLE_NAME：指定
cust_parquet或cust_orc作为 Hive 表名称。
预期输出代码段：
- TABLE_NAME：指定
...
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}