Spark Structured Streaming样例程序开发思路

场景说明

在Spark应用中，通过使用StructuredStreaming调用Kafka接口来获取单词记录，然后把单词记录分类统计，得到每个单词记录数。

数据规划

开发思路

1. 接收Kafka中数据，生成相应DataStreamReader。
2. 对单词记录进行分类统计。
3. 计算结果，并进行打印。

运行前置操作

安全模式下Spark Core样例代码需要读取两个文件（user.keytab、krb5.conf）。user.keytab和krb5.conf文件为安全模式下的认证文件，需要在FusionInsight Manager中下载principal用户的认证凭证，样例代码中使用的用户为：sparkuser，需要修改为准备好的开发用户。

打包项目

• 将user.keytab、krb5.conf 两个文件上传客户端所在服务器上。

• 通过IDEA自带的Maven工具，打包项目，生成jar包。具体操作请参考在Linux环境中调测Spark应用。
  

  编译打包前，样例代码中的user.keytab、krb5.conf文件路径需要修改为该文件所在客户端服务器的实际路径。例如：“/opt/female/user.keytab”，“/opt/female/krb5.conf”。

• 将打包生成的jar包上传到Spark客户端所在服务器的任意目录（例如“ /opt” ）下。
• 将commons-pool2-xxx.jar上传到“$SPARK_HOME/jars/streamingClient010/”目录下（jar包可从$SPARK_HOME/tool/carbonPrequery目录下获取）。

运行任务

• 在运行样例程序时需要指定<brokers> <subscribe-type> <topic> <protocol> <service> <domain><checkpointDir>，其中<brokers>指获取元数据的Kafka地址（需使用21007端口），<subscribe-type>指Kafka订阅类型（如subscribe），<topic>指读取Kafka上的topic名称，<protocol>指安全访问协议（如SASL_PLAINTEXT），<service>指kerberos服务名称（如kafka），<domain>指kerberos域名（如hadoop.<系统域名>），<checkpointDir>指checkpoint文件存放路径。
  

  • 由于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[*]}")
    由于运行模式为安全模式，需要添加新配置并修改命令参数：

    1. $SPARK_HOME/conf/jaas.conf添加新配置：

       KafkaClient {
       com.sun.security.auth.module.Krb5LoginModule required
       useKeyTab=false
       useTicketCache=true
       debug=false;
       };

    2. $SPARK_HOME/conf/jaas-zk.conf添加新配置：

       KafkaClient {
       com.sun.security.auth.module.Krb5LoginModule required
       useKeyTab=true
       keyTab="./user.keytab"
       principal="sparkuser@<系统域名>"
       useTicketCache=false
       storeKey=true
       debug=true;
       };

    3. 使用--files和相对路径提交keytab文件，这样才能保证keytab文件被加载到executor的container中。
  • 用户提交结构流任务时，通常需要通过--jars命令指定kafka相关jar包的路径，当前版本用户除了这一步外还需要将$SPARK_HOME/jars/streamingClient010目录中的kafka-clients jar包复制到$SPARK_HOME/jars目录下，否则会报class not found异常。