# Spark Structured Streaming对接Kafka样例程序开发思路
#### 场景说明
假定一个广告业务，存在广告请求事件、广告展示事件、广告点击事件，广告主需要实时统计有效的广告展示和广告点击数据。
已知：
1. 终端用户每次请求一个广告后，会生成广告请求事件，保存到kafka的adRequest topic中。
2. 请求一个广告后，可能用于多次展示，每次展示，会生成广告展示事件，保存到kafka的adShow topic中。
3. 每个广告展示，可能会产生多次点击，每次点击，会生成广告点击事件，保存到kafka的adClick topic中。
4. 广告有效展示的定义如下：
   1. 请求到展示的时长超过A分钟算无效展示。
   
   2. A分钟内多次展示，每次展示事件为有效展示。
    
5. 广告有效点击的定义如下：
   1. 展示到点击时长超过B分钟算无效点击。
   
   2. B分钟内多次点击，仅首次点击事件为有效点击。
    
基于此业务场景，模拟简单的数据结构如下：
- 广告请求事件 数据结构：adID\^reqTime
  
- 广告展示事件 数据结构：adID\^showID\^showTime
  
- 广告点击事件 数据结构：adID\^showID\^clickTime
  
数据关联关系如下：
- 广告请求事件与广告展示事件通过adID关联。
- 广告展示事件与广告点击事件通过adID+showID关联。
数据要求：
- 数据从产生到到达流处理引擎的延迟时间不超过2小时
- 广告请求事件、广告展示事件、广告点击事件到达流处理引擎的时间不能保证有序和时间对齐
 
#### 数据规划
1. 在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}***
   ![](https://support.huaweicloud.com/devg3-mrs/public_sys-resources/note_3.0-zh-cn.png)
   - 确保集群安装完成，包括HDFS、Yarn、Spark2x和Kafka。
   
   - 启动Kafka的Producer，向Kafka发送数据。
   
   - {ClassPath}表示工程jar包的存放路径，详细路径由用户指定，可参考[在Linux环境中编包并运行Spark程序](https://support.huaweicloud.com/devg3-mrs/mrs_07_410123.html)章节中导出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 }
   
 
#### 开发思路
1. 使用Structured Streaming接收Kafka中数据，生成请求流、展示流、点击流。
2. 对请求流、展示流、点击流的数据进行关联查询。
3. 统计结果写入kafka。
4. 应用中监控流处理任务的状态。
 
#### 打包项目
- 通过IDEA自带的Maven工具，打包项目，生成jar包。具体操作请参考[在Linux环境中编包并运行Spark程序](https://support.huaweicloud.com/devg3-mrs/mrs_07_410123.html)。
- 将打包生成的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路径下。
![](https://support.huaweicloud.com/devg3-mrs/public_sys-resources/note_3.0-zh-cn.png)
由于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\[\*\]}")**
![](https://support.huaweicloud.com/devg3-mrs/public_sys-resources/caution_3.0-zh-cn.png)
用户提交结构流任务时，通常需要通过**--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\>*