Spark Structured Streaming对接Kafka样例程序开发思路

场景说明

假定一个广告业务，存在广告请求事件、广告展示事件、广告点击事件，广告主需要实时统计有效的广告展示和广告点击数据。

已知：

终端用户每次请求一个广告后，会生成广告请求事件，保存到kafka的adRequest topic中。
请求一个广告后，可能用于多次展示，每次展示，会生成广告展示事件，保存到kafka的adShow topic中。
每个广告展示，可能会产生多次点击，每次点击，会生成广告点击事件，保存到kafka的adClick topic中。
广告有效展示的定义如下：
1. 请求到展示的时长超过A分钟算无效展示。
2. A分钟内多次展示，每次展示事件为有效展示。
广告有效点击的定义如下：
1. 展示到点击时长超过B分钟算无效点击。
2. B分钟内多次点击，仅首次点击事件为有效点击。

基于此业务场景，模拟简单的数据结构如下：

广告请求事件
 数据结构：adID^reqTime
广告展示事件
 数据结构：adID^showID^showTime
广告点击事件
 数据结构：adID^showID^clickTime

数据关联关系如下：

广告请求事件与广告展示事件通过adID关联。
广告展示事件与广告点击事件通过adID+showID关联。

数据要求：

数据从产生到到达流处理引擎的延迟时间不超过2小时
广告请求事件、广告展示事件、广告点击事件到达流处理引擎的时间不能保证有序和时间对齐

数据规划

在kafka中生成模拟数据（需要有Kafka权限用户）。
java -cp $SPARK_HOME/conf:$SPARK_HOME/jars/*:$SPARK_HOME/jars/streamingClient010/*:{ClassPath} com.huawei.bigdata.spark.examples.KafkaADEventProducer {BrokerList} {timeOfProduceReqEvent} {eventTimeBeforeCurrentTime} {reqTopic} {reqEventCount} {showTopic} {showEventMaxDelay} {clickTopic} {clickEventMaxDelay}
- 确保集群安装完成，包括HDFS、Yarn、Spark2x和Kafka。
- 启动Kafka的Producer，向Kafka发送数据。
- {ClassPath}表示工程jar包的存放路径，详细路径由用户指定，可参考在Linux环境中编包并运行Spark程序章节中导出jar包的操作步骤。
命令举例：

java -cp /opt/client/Spark2x/spark/conf:/opt/StructuredStreamingADScalaExample-1.0.jar:/opt/client/Spark2x/spark/jars/*:/opt/client/Spark2x/spark/jars/streamingClient010/* com.huawei.bigdata.spark.examples.KafkaADEventProducer 10.132.190.170:21005,10.132.190.165:21005 2h 1h req 10000000 show 5m click 5m

此命令将在kafka上创建3个topic：req、show、click，在2h内生成1千万条请求事件数据，请求事件的时间取值范围为{当前时间-1h 至当前时间}，并为每条请求事件随机生成0-5条展示事件，展示事件的时间取值范围为{请求事件时间至请求事件时间+5m }，为每条展示事件随机生成0-5条点击事件，点击事件的时间取值范围为{展示事件时间至展示事件时间+5m }

开发思路

使用Structured Streaming接收Kafka中数据，生成请求流、展示流、点击流。
对请求流、展示流、点击流的数据进行关联查询。
统计结果写入kafka。
应用中监控流处理任务的状态。

打包项目

通过IDEA自带的Maven工具，打包项目，生成jar包。具体操作请参考在Linux环境中编包并运行Spark程序。
将打包生成的jar包上传到Spark客户端所在服务器的任意目录（例如“ /opt” ）下。
将user.keytab、krb5.conf 两个文件上传客户端所在服务器上

运行任务

在运行样例程序时需要指定 <kafkaBootstrapServers> <maxEventDelay> <reqTopic> <showTopic> <maxShowDelay> <clickTopic> <maxClickDelay> <triggerInterver> <checkpointDir>。

<kafkaBootstrapServers>指获取元数据的Kafka地址。
<maxEventDelay>指数据从生成到被流处理引擎的最大延迟时间。
<reqTopic>指请求事件的topic名称。
<showTopic>指展示事件的topic名称。
<maxShowDelay>指有效展示事件的最大延迟时间。
<clickTopic>指点击事件的topic名称。
<maxClickDelay>指有效点击事件的最大延迟时间。
<triggerInterver>指流处理任务的触发间隔。
<checkpointDir>指checkpoint文件存放路径，本地或者HDFS路径下。

由于Spark Structured Streaming Kafka的依赖包在客户端的存放路径与其他依赖包不同，如其他依赖包路径为“$SPARK_HOME/jars”，而Spark Structured Streaming Kafka依赖包路径为“$SPARK_HOME/jars/streamingClient010”。所以在运行应用程序时，需要在spark-submit命令中添加配置项，指定Spark Streaming Kafka的依赖包路径，如--jars $(files=($SPARK_HOME/jars/streamingClient010/*.jar); IFS=,; echo "${files[*]}")

用户提交结构流任务时，通常需要通过--jars命令指定kafka相关jar包的路径，当前版本用户除了这一步外还需要将$SPARK_HOME/jars/streamingClient010目录中的kafka-clients jar包复制到$SPARK_HOME/jars目录下，否则会报class not found异常。

进入Spark客户端目录，调用bin/spark-submit脚本运行代码，运行命令如下（类名与文件名等请与实际代码保持一致，此处仅为示例）：

bin/spark-submit --master yarn --deploy-mode client --jars $(files=($SPARK_HOME/jars/streamingClient010/*.jar); IFS=,; echo "${files[*]}") --conf "spark.sql.streaming.statefulOperator.checkCorrectness.enabled=false" --class com.huawei.bigdata.spark.examples.KafkaADCount /opt/StructuredStreamingADScalaExample-1.0.jar <kafkaBootstrapServers> <maxEventDelay> <reqTopic> <showTopic> <maxShowDelay> <clickTopic> <maxClickDelay> <triggerInterver> <checkpointDir>

父主题： Spark Structured Streaming对接Kafka样例程序

上一篇：Spark Structured Streaming对接Kafka样例程序

下一篇：Spark Structured Streaming对接Kafka样例程序（Scala）