场景说明
场景说明
假定一个广告业务，存在广告请求事件、广告展示事件、广告点击事件，广告主需要实时统计有效的广告展示和广告点击数据。
已知：
- 终端用户每次请求一个广告后，会生成广告请求事件，保存到kafka的adRequest topic中。
- 请求一个广告后，可能用于多次展示，每次展示，会生成广告展示事件，保存到kafka的adShow topic中。
- 每个广告展示，可能会产生多次点击，每次点击，会生成广告点击事件，保存到kafka的adClick topic中。
- 广告有效展示的定义如下：
- 请求到展示的时长超过A分钟算无效展示。
- A分钟内多次展示，每次展示事件为有效展示。
- 广告有效点击的定义如下：
- 展示到点击时长超过B分钟算无效点击。
- B分钟内多次点击，仅首次点击事件为有效点击。
基于此业务场景，模拟简单的数据结构如下：
数据关联关系如下：
- 广告请求事件与广告展示事件通过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的Broker配置参数“allow.everyone.if.no.acl.found”的值修改为“true”。
- 启动Kafka的Producer，向Kafka发送数据。
- {ClassPath}表示工程jar包的存放路径，详细路径由用户指定，可参考编包并运行程序章节中导出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包。具体操作请参考编包并运行程序。
- 将打包生成的jar包上传到Spark客户端所在服务器的任意目录（例如“ /opt” ）下。
- 将user.keytab、krb5.conf 两个文件上传客户端所在服务器上
运行任务
在运行样例程序时需要指定 <kafkaBootstrapServers> <maxEventDelay> <reqTopic> <showTopic> <maxShowDelay> <clickTopic> <maxClickDelay> <triggerInterver> <checkpointDir>。<checkpointDir>指checkpoint文件存放路径。
- <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>
