通过Spark Streaming作业处理Kafka数据

前提条件

已开通E-MapReduce服务。
已完成云账号的授权，详情请参见角色授权。

步骤一：创建DataLake和DataFlow集群

创建同一个安全组下的DataLake和DataFlow集群（包含Kafka服务）。创建详情请参见创建集群。

创建DataLake集群。

说明
本文以Spark 3为例。

在软件配置步骤中，地域选择华东1（杭州），业务场景选择数据湖，产品版本选择EMR-5.17.3 正式版本，服务高可用保持关闭。在可选服务区域中选中HADOOP-COMMON(3.2.1)、HDFS(3.2.1)、OSS-HDFS(1.0.0)、HIVE(3.1.3)、SPARK2(2.4.8)、SPARK3(3.4.2)、TEZ(0.10.2)、TRINO(422)、Deltalake(3.0.0)、YARN(3.2.1)等组件。
创建DataFlow。

说明
务必确认已选择Kafka服务。系统会自动选择其依赖的Kafka-Manager和Zookeeper服务。

在业务场景区域选择实时数据流，产品版本选择 EMR-5.17.3 正式版本，在可选服务区域确认 Kafka、Kafka-Manager 和 ZOOKEEPER 已选中。

步骤二：获取JAR包并上传到DataLake集群

获取JAR包（spark-streaming-demo-1.0.jar）。
上传JAR包至DataLake集群Master节点的/home/emr-user路径下。

步骤三：在DataFlow集群上创建Topic

本示例将创建一个名称为test的Topic。

登录DataFlow集群的Master节点，详情请参见登录集群。

执行以下命令，创建Topic。

kafka-topics.sh --create --bootstrap-server core-1-1:9092,core-1-2:9092,core-1-3:9092 --replication-factor 2 --partitions 2 --topic demo

说明

创建Topic后，请保留该登录窗口，后续步骤仍将使用。

步骤四：运行Spark Streaming作业

本示例将运行一个流式单词统计（WordCount）的作业。

登录DataLake集群的Master节点，详情请参见登录集群。
执行以下命令，进入emr-user目录。
```
cd /home/emr-user
```

执行以下命令，进行流式单词统计（WordCount）。

spark-submit --class com.aliyun.emr.KafkaApp1 ./spark-streaming-demo-1.0.jar <Kafka broker的内网IP地址>:9092 demogroup1 demo

关键参数如下表所示。

参数	描述
`<Kafka broker的内网IP地址>:9092`	DataFlow集群中Broker节点的内网IP地址和端口号，端口号默认为9092。例如，`172.16..:9092,172.16..:9092,172.16..:9092`。您可以在DataFlow集群Kafka服务的状态页签，单击KafkaBroker组件前的图标，查看所有节点的内网IP地址。
`demogroup1`	指定Kafka消费组的名称。您也可以根据实际情况修改。
`demo`	Topic名称。

步骤五：使用Kafka发布消息

在DataFlow集群的命令行窗口，执行如下命令运行Kafka的生产者。

kafka-console-producer.sh --topic demo --bootstrap-server core-1-1:9092,core-1-2:9092,core-1-3:9092

在DataFlow集群的登录窗口中输入文本，在DataLake集群的登录窗口中，会实时显示文本的统计信息。

例如，在DataFlow集群的登录窗口输入信息。

kafka-console-producer.sh --topic demo --bootstrap-server core-1-1:9092,core-1-2:9092,core-1-3:9092
>hello

DataLake集群的登录窗口会输出如下信息。

24/01/11 20:33:26 INFO [dag-scheduler-event-loop] DAGScheduler: Missing parents: List()
24/01/11 20:33:26 INFO [dag-scheduler-event-loop] DAGScheduler: Submitting ResultStage 95 (ShuffledRDD[119] at reduceByKey at KafkaApp1.scala:45), which has no missing parents
24/01/11 20:33:26 INFO [dag-scheduler-event-loop] MemoryStore: Block broadcast_71 stored as values in memory (estimated size 4.8 KiB, free 912.2 MiB)
24/01/11 20:33:26 INFO [dag-scheduler-event-loop] MemoryStore: Block broadcast_71_piece0 stored as bytes in memory (estimated size 2.8 KiB, free 912.2 MiB)
24/01/11 20:33:26 INFO [dispatcher-BlockManagerMaster] BlockManagerInfo: Added broadcast_71_piece0 in memory on master-1-1.c-10046faxxx.cn-hangzhou.emr.aliyuncs.com:45923 (size: 2.8 KiB, free: 912.3 MiB)
24/01/11 20:33:26 INFO [dag-scheduler-event-loop] SparkContext: Created broadcast 71 from broadcast at DAGScheduler.scala:1520
24/01/11 20:33:26 INFO [dag-scheduler-event-loop] DAGScheduler: Submitting 1 missing tasks from ResultStage 95 (ShuffledRDD[119] at reduceByKey at KafkaApp1.scala:45) (first 15 tasks are for partitions Vector(1))
24/01/11 20:33:26 INFO [dag-scheduler-event-loop] YarnScheduler: Adding task set 95.0 with 1 tasks resource profile 0
24/01/11 20:33:26 INFO [dispatcher-CoarseGrainedScheduler] TaskSetManager: Starting task 0.0 in stage 95.0 (TID 95) (core-1-2.c-10046faaxxx.cn-hangzhou.emr.aliyuncs.com, executor 1, partition 1, PROCESS_LOCAL, 4282 bytes) taskResourceAssignments Map()
24/01/11 20:33:26 INFO [dispatcher-BlockManagerMaster] BlockManagerInfo: Added broadcast_71_piece0 in memory on core-1-2.c-10046faxxx-hangzhou.emr.aliyuncs.com:40147 (size: 2.8 KiB, free: 1418.1 MiB)
24/01/11 20:33:26 INFO [task-result-getter-3] TaskSetManager: Finished task 0.0 in stage 95.0 (TID 95) in 45 ms on core-1-2.c-100xxx.cn-hangzhou.emr.aliyuncs.com (executor 1) (1/1)
24/01/11 20:33:26 INFO [task-result-getter-3] YarnScheduler: Removed TaskSet 95.0, whose tasks have all completed, from pool
24/01/11 20:33:26 INFO [dag-scheduler-event-loop] DAGScheduler: ResultStage 95 (print at KafkaApp1.scala:46) finished in 0.066 s
24/01/11 20:33:26 INFO [dag-scheduler-event-loop] DAGScheduler: Job 47 is finished. Cancelling potential speculative or zombie tasks for this job
24/01/11 20:33:26 INFO [dag-scheduler-event-loop] YarnScheduler: Killing all running tasks in stage 95: Stage finished
24/01/11 20:33:26 INFO [streaming-job-executor-0] DAGScheduler: Job 47 finished: print at KafkaApp1.scala:46, took 0.069926 s
-------------------------------------------
Time: 1704976406000 ms
-------------------------------------------
(hello,1)

步骤六：查看Spark Streaming作业状态

在EMR on ECS页面，单击DataLake集群的名称。
单击访问链接与端口页签。
访问SPARK UI，详情请参见通过控制台访问开源组件Web界面。
在History Server页面，单击待查看的App ID。

您可以查看Spark Streaming作业的状态。

在 Spark Web UI 的 Jobs 页面，可查看应用 DirectKafkaWordCount 的流式作业执行状态。页面显示 Scheduling Mode 为 FIFO，Completed Jobs 共 409 个。作业列表包含 Job Id、Description、Submitted、Duration、Stages: Succeeded/Total、Tasks: Succeeded/Total 等列，可据此确认各批次流式作业的完成状态与耗时。