# Spark Structured Streaming状态操作样例程序开发思路
#### 场景说明
假设需要跨批次统计每个session期间发生了多少次event以及本session的开始和结束timestamp；
同时输出本批次被更新状态的session。
#### 数据规划
1. 在kafka中生成模拟数据（需要有Kafka权限用户）
2. 确保集群安装完成，包括HDFS、Yarn、Spark2x和Kafka。
3. 创建Topic。 {zkQuorum}表示ZooKeeper集群信息，格式为IP:port。
   *$KAFKA_HOME/bin/kafka-topics.sh --create --zookeeper {zkQuorum}/kafka --replication-factor 1 --partitions 1 --topic {Topic}*
   
4. 启动Kafka的Producer，向Kafka发送数据。 {ClassPath}表示工程jar包的存放路径，详细路径由用户指定，可参考[在Linux环境中编包并运行Spark程序](https://support.huaweicloud.com/devg3-mrs/mrs_07_410123.html)章节中导出jar包的操作步骤。
   java -cp $SPARK_HOME/conf:$SPARK_HOME/jars/\*:$SPARK_HOME/jars/streamingClient010/\*:{ClassPath} com.huawei.bigdata.spark.examples.KafkaProducer {brokerlist} {topic} {number of events produce every 0.02s}
   示例：
   java -cp /opt/client/Spark2x/spark/conf:/opt/StructuredStreamingState-1.0.jar:/opt/client/Spark2x/spark/jars/\*:/opt/client/Spark2x/spark/jars/streamingClient010/\* com.huawei.bigdata.spark.examples.KafkaProducer xxx.xxx.xxx.xxx:21005,xxx.xxx.xxx.xxx:21005,xxx.xxx.xxx.xxx:21005 mytopic 10
   
 
#### 开发思路
1. 接收Kafka中数据，生成相应DataStreamReader。
2. 进行分类统计。
3. 计算结果，并进行打印。
 
#### 打包项目
- 通过IDEA自带的Maven工具，打包项目，生成jar包
- 将打包生成的jar包上传到Spark客户端所在服务器的任意目录（例如" /opt" ）下。
 
#### 运行任务
在运行样例程序时需要指定以下参数：
- brokers：获取元数据的Kafka地址。
- subscribe-type：指定Kafka的消费方式。
- topic：指要消费的Kafka topic。
- checkpointLocation：指Spark任务的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 Streaming Structured Kafka依赖包路径为"$SPARK_HOME/jars/streamingClient010"。所以在运行应用程序时，需要在spark-submit命令中添加配置项，指定Spark Streaming Kafka的依赖包路径，如**--jars $(files=($SPARK_HOME/jars/streamingClient010/\*.jar); IFS=,; echo "${files\[\*\]}")**
进入Spark客户端目录，调用bin/spark-submit脚本运行代码，运行命令如下（类名与文件名等请与实际代码保持一致，此处仅为示例）：
- **bin/spark-submit --master yarn --deploy-mode client** ***--jars $(files=($SPARK_HOME/jars/streamingClient010/\*.jar); IFS=,; echo "${files\[\*\]}")*** **--class** *com.huawei.bigdata.spark.examples.kafkaSessionization /opt/StructuredStreamingState-1.0.jar \<brokers\> \<subscribe-type\>* \<topic\> \<checkpointLocation\>
  ![](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异常。