华为云用户手册

打包项目 通过IDEA自带的Maven工具，打包项目，生成jar包。具体操作请参考在Linux环境中调测Spark应用。 将打包生成的jar包上传到Spark客户端所在服务器的任意目录（例如“$SPARK_HOME” ）下。 将user.keytab、krb5.conf 两个文件上传客户端所在服务器上（文件上传的路径需要和生成的jar包路径一致）。 若运行“Spark on HBase”样例程序，需要在Spark客户端的“spark-defaults.conf”配置文件中将配置项“spark.yarn.security.credentials.hbase.enabled”设置为“true”（该参数值默认为“false”，改为“true”后对已有业务没有影响。如果要卸载HBase服务，卸载前请将此参数值改回“false”），将配置项“spark.inputFormat.cache.enabled”设置为“false”。

Python样例代码 下面代码片段仅为演示，具体代码参见SparkOnHbasePythonExample中HBaseDistributedScanExample文件： # -*- coding:utf-8 -*-# -*- coding:utf-8 -*-"""【说明】(1)由于pyspark不提供Hbase相关api,本样例使用Python调用Java的方式实现(2)如果使用yarn-client模式运行,请确认Spark2x客户端Spark2x/spark/conf/spark-defaults.conf中 spark.yarn.security.credentials.hbase.enabled参数配置为true"""from py4j.java_gateway import java_importfrom pyspark.sql import SparkSession# 创建SparkSessionspark = SparkSession\ .builder\ .appName("JavaHBaseDistributedScan")\ .getOrCreate()# 向sc._jvm中导入要运行的类java_import(spark._jvm, 'com.huawei.bigdata.spark.examples.hbasecontext.JavaHBaseDistributedScanExample')# 创建类实例并调用方法，传递sc._jsc参数spark._jvm.JavaHBaseDistributedScan().execute(spark._jsc, sys.argv)# 停止SparkSessionspark.stop()

回答 原因分析：显示该异常是因为“recoverFromCheckpointLocation”的值判定为false，但却配置了checkpoint目录。 参数“recoverFromCheckpointLocation”的值为代码中“outputMode == OutputMode.Complete()”语句的判断结果（outputMode的默认输出方式为“append”）。 处理方法：编写应用时，用户可以根据具体情况修改数据的输出方式。 将输出方式修改为“complete”，“recoverFromCheckpointLocation”的值会判定为true。此时配置了checkpoint目录时就不会显示异常。

回答 如下图所示，Spark Streaming应用中定义的逻辑为，从Kafka中读取数据，执行对应处理之后，然后将结果数据回写至Kafka中。 例如：Spark Streming中定义了批次时间，如果数据传入Kafka的速率为10MB/s，而Spark Streaming中定义了每60s一个批次，回写数据总共为600MB。而Kafka中定义了接收数据的阈值大小为500MB。那么此时回写数据已超出阈值。此时，会出现上述错误。 图1 应用场景 解决措施： 方式一：推荐优化Spark Streaming应用程序中定义的批次时间，降低批次时间，可避免超过Kafka定义的阈值。一般建议以5-10秒/次为宜。 方式二：将Kafka的阈值调大，建议在 FusionInsight Manager中的Kafka服务进行参数设置，将socket.request.max.bytes参数值根据应用场景，适当调整。

回答 Spark任务在运行过程中，driver会创建一个spark-开头的本地临时目录，用于存放业务jar包，配置文件等，同时在本地创建一个blockmgr-开头的本地临时目录，用于存放block data。此两个目录会在Spark应用运行结束时自动删除。 此两个目录的存放路径优先通过SPARK_LOCAL_DIRS环境变量指定，若不存在该环境变量，则设置为spark.local.dir的值，若此配置还不存在，则使用java.io.tmpdir的值。客户端默认配置中spark.local.dir被设置为/tmp，因此默认使用系统/tmp目录。 但存在一些特殊情况，如driver进程未正常退出，比如被kill -9命令结束进程，或者Java虚拟机直接崩溃等场景，导致driver的退出流程未正常执行，则可能导致该部分目录无法被正常清理，残留在系统中。 当前只有yarn-client模式和local模式的driver进程会产生上述问题，在yarn-cluster模式中，已将container内进程的临时目录设置为container临时目录，当container退出时，由container自动清理该目录，因此yarn-cluster模式不存在此问题。

Python样例代码 下面代码片段仅为演示，具体代码参见SparkOnHbasePythonExample中HBaseBulkDeleteExample文件： def main(args: Array[String]) {# -*- coding:utf-8 -*-"""【说明】(1)由于pyspark不提供Hbase相关api,本样例使用Python调用Java的方式实现(2)如果使用yarn-client模式运行,请确认Spark2x客户端Spark2x/spark/conf/spark-defaults.conf中 spark.yarn.security.credentials.hbase.enabled参数配置为true"""from py4j.java_gateway import java_importfrom pyspark.sql import SparkSession# 创建SparkSessionspark = SparkSession\ .builder\ .appName("JavaHBaseBulkDeleteExample")\ .getOrCreate()# 向sc._jvm中导入要运行的类java_import(spark._jvm, 'com.huawei.bigdata.spark.examples.hbasecontext.JavaHBaseBulkDeleteExample')# 创建类实例并调用方法，传递sc._jsc参数spark._jvm.JavaHBaseBulkDeleteExample().execute(spark._jsc, sys.argv)# 停止SparkSessionspark.stop()

打包项目 通过IDEA自带的Maven工具，打包项目，生成jar包。具体操作请参考在Linux环境中调测Spark应用。 将打包生成的jar包上传到Spark客户端所在服务器的任意目录（例如“$SPARK_HOME” ）下。 将user.keytab、krb5.conf 两个文件上传客户端所在服务器上（文件上传的路径需要和生成的jar包路径一致）。 若运行“Spark on HBase”样例程序，需要在Spark客户端的“spark-defaults.conf”配置文件中将配置项“spark.yarn.security.credentials.hbase.enabled”设置为“true”（该参数值默认为“false”，改为“true”后对已有业务没有影响。如果要卸载HBase服务，卸载前请将此参数值改回“false”），将配置项“spark.inputFormat.cache.enabled”设置为“false”。

回答 Spark部署时，如下jar包存放在客户端的“${SPARK_HOME}/jars/streamingClient010”目录以及服务端的“${BIGDATA_HOME}/FusionInsight_Spark2x_8.1.0.1/install/FusionInsight-Spark2x-3.1.1/spark/jars/streamingClient010”目录： kafka-clients-xxx.jar kafka_2.12-xxx.jar spark-streaming-kafka-0-10_2.12-3.1.1-hw-ei-311001-SNAPSHOT.jar spark-token-provider-kafka-0-10_2.12-3.1.1-hw-ei-311001-SNAPSHOT.jar 由于“$SPARK_HOME/jars/streamingClient010/*”默认没有添加到classpath，所以需要手动配置。 在提交应用程序运行时，在命令中添加如下参数即可，详细示例可参考在Linux环境中调测Spark应用。 --jars $SPARK_CLIENT_HOME/jars/streamingClient010/kafka-client-2.4.0.jar,$SPARK_CLIENT_HOME/jars/streamingClient010/kafka_2.12-2.4.0.jar,$SPARK_CLIENT_HOME/jars/streamingClient010/spark-streaming-kafka-0-10_2.12-3.1.1-hw-ei-311001-SNAPSHOT.jar 用户自己开发的应用程序以及样例工程都可使用上述命令提交。 但是Spark开源社区提供的KafkaWordCount等样例程序，不仅需要添加--jars参数，还需要配置其他，否则会报“ClassNotFoundException”错误，yarn-client和yarn-cluster模式下稍有不同。 yarn-client模式下 在除--jars参数外，在客户端“spark-defaults.conf”配置文件中，将“spark.driver.extraClassPath”参数值中添加客户端依赖包路径，如“$SPARK_HOME/jars/streamingClient010/*”。 yarn-cluster模式下 除--jars参数外，还需要配置其他，有三种方法任选其一即可，具体如下： 在客户端spark-defaults.conf配置文件中，在“spark.yarn.cluster.driver.extraClassPath”参数值中添加服务端的依赖包路径，如“${BIGDATA_HOME}/FusionInsight_Spark2x_8.1.0.1/install/FusionInsight-Spark2x-3.1.1/spark/jars/streamingClient010/*”。 将各服务端节点的“original-spark-examples_2.12-3.1.1-xxx.jar”包删除。 在客户端“spark-defaults.conf”配置文件中，修改或增加配置选项“spark.driver.userClassPathFirst” = “true”。

问题 使用运行的Spark Streaming任务回写Kafka时，Kafka上接收不到回写的数据，且Kafka日志报错信息如下： 2016-03-02 17:46:19,017 | INFO | [kafka-network-thread-21005-1] | Closing socket connection to /10.91.8.208 due to invalid request: Request of length 122371301 is not valid, it is larger than the maximum size of 104857600 bytes. | kafka.network.Processor (Logging.scala:68)2016-03-02 17:46:19,155 | INFO | [kafka-network-thread-21005-2] | Closing socket connection to /10.91.8.208. | kafka.network.Processor (Logging.scala:68)2016-03-02 17:46:19,270 | INFO | [kafka-network-thread-21005-0] | Closing socket connection to /10.91.8.208 due to invalid request: Request of length 122371301 is not valid, it is larger than the maximum size of 104857600 bytes. | kafka.network.Processor (Logging.scala:68)2016-03-02 17:46:19,513 | INFO | [kafka-network-thread-21005-1] | Closing socket connection to /10.91.8.208 due to invalid request: Request of length 122371301 is not valid, it is larger than the maximum size of 104857600 bytes. | kafka.network.Processor (Logging.scala:68)2016-03-02 17:46:19,763 | INFO | [kafka-network-thread-21005-2] | Closing socket connection to /10.91.8.208 due to invalid request: Request of length 122371301 is not valid, it is larger than the maximum size of 104857600 bytes. | kafka.network.Processor (Logging.scala:68)53393 [main] INFO org.apache.hadoop.mapreduce.Job - Counters: 50

提交命令 假设用例代码打包后的jar包名为spark-hbaseContext-test-1.0.jar，并将jar包放在客户端“$SPARK_HOME”目录下，以下命令均在“$SPARK_HOME”目录执行，Java接口对应的类名前有Java字样，请参考具体样例代码进行书写。 yarn-client模式： java/scala 版本（类名等请与实际代码保持一致，此处仅为示例） bin/spark-submit --master yarn --deploy-mode client --class com.huawei.bigdata.spark.examples.hbasecontext.JavaHBaseBulkDeleteExample SparkOnHbaseJavaExample.jar bulktable python版本（文件名等与实际保持一致，此处仅为示例），假设对应的Java代码打包后包名为SparkOnHbaseJavaExample.jar，且放在当前提交目录。 bin/spark-submit --master yarn --deploy-mode client --jars SparkOnHbaseJavaExample.jar HBaseButDeleteExample.py bulktable yarn-cluster模式： java/scala 版本（类名等请与实际代码保持一致，此处仅为示例） bin/spark-submit --master yarn --deploy-mode cluster --class com.huawei.bigdata.spark.examples.hbasecontext.JavaHBaseBulkDeleteExample --files /opt/user.keytab,/opt/krb5.conf SparkOnHbaseJavaExample.jar bulktable python版本（文件名等与实际保持一致，此处仅为示例），假设对应的Java代码打包后包名为SparkOnHbaseJavaExample.jar，且放在当前提交目录。 bin/spark-submit --master yarn --deploy-mode cluster --files /opt/user.keytab,/opt/krb5.conf --jars SparkOnHbaseJavaExample.jar HBaseButDeleteExample.py bulktable

打包项目 将user.keytab、krb5.conf 两个文件上传客户端所在服务器上。 通过IDEA自带的Maven工具，打包项目，生成jar包。具体操作请参考在Linux环境中调测Spark应用。 编译打包前，样例代码中的user.keytab、krb5.conf文件路径需要修改为该文件所在客户端服务器的实际路径。例如：“/opt/female/user.keytab”，“/opt/female/krb5.conf”。 将打包生成的jar包上传到Spark客户端所在服务器的任意目录（例如“ /opt” ）下。 准备依赖包，将下列jar包上传到Spark客户端所在服务器，“$SPARK_HOME/jars/streamingClient010”目录下。 spark-streaming-kafkaWriter-0-10_2.12-3.1.1-hw-ei-311001.jar kafka-clients-xxx.jar kafka_2.12-xxx.jar spark-sql-kafka-0-10_2.12-3.1.1-hw-ei-311001-SNAPSHOT.jar spark-streaming-kafka-0-10_2.12-3.1.1-hw-ei-311001-SNAPSHOT.jar spark-token-provider-kafka-0-10_2.12-3.1.1-hw-ei-311001-SNAPSHOT.jar 版本号中包含hw-ei的依赖包请从华为开源镜像站下载。 版本号中不包含hw-ei的依赖包都来自开源仓库，请从Maven中心仓获取。

数据规划 Spark Streaming样例工程的数据存储在Kafka组件中。向Kafka组件发送数据（需要有Kafka权限用户）。 确保集群安装完成，包括HDFS、Yarn、Spark和Kafka。 本地新建文件“input_data1.txt”，将“log1.txt”的内容复制保存到“input_data1.txt”。 在客户端安装节点下创建文件目录：“/home/data”。将上述文件上传到此“/home/data”目录下。 将Kafka的Broker配置参数“allow.everyone.if.no.acl.found”的值修改为“true”。 创建Topic。 {zkQuorum}表示ZooKeeper集群信息，格式为IP:port。 $KAFKA_HOME/bin/kafka-topics.sh --create --zookeeper {zkQuorum}/kafka --replication-factor 1 --partitions 3 --topic {Topic} 启动Kafka的Producer，向Kafka发送数据。 java -cp {ClassPath} com.huawei.bigdata.spark.examples.StreamingExampleProducer {BrokerList} {Topic} 其中，ClassPath应包含Spark客户端Kafka jar包的绝对路径，如/opt/client/Spark2x/spark/jars/*:/opt/client/Spark2x/spark/jars/streamingClient010/*

代码样例 由于pyspark不提供Hbase相关api，本样例使用Python调用Java的方式实现。 下面代码片段仅为演示，具体代码参见SparkHivetoHbasePythonExample: # -*- coding:utf-8 -*-from py4j.java_gateway import java_importfrom pyspark.sql import SparkSession# 创建SparkSessionspark = SparkSession\ .builder\ .appName("SparkHivetoHbase") \ .getOrCreate()# 向sc._jvm中导入要运行的类java_import(spark._jvm, 'com.huawei.bigdata.spark.examples.SparkHivetoHbase')# 创建类实例并调用方法spark._jvm.SparkHivetoHbase().hivetohbase(spark._jsc)# 停止SparkSessionspark.stop()

打包项目 通过IDEA自带的Maven工具，打包项目，生成jar包。具体操作请参考在Linux环境中调测Spark应用。 将打包生成的jar包上传到Spark客户端所在服务器的任意目录（例如“$SPARK_HOME” ）下。 将user.keytab、krb5.conf 两个文件上传客户端所在服务器上（文件上传的路径需要和生成的jar包路径一致）。 若运行“Spark on HBase”样例程序，需要在Spark客户端的“spark-defaults.conf”配置文件中将配置项“spark.yarn.security.credentials.hbase.enabled”设置为“true”（该参数值默认为“false”，改为“true”后对已有业务没有影响。如果要卸载HBase服务，卸载前请将此参数值改回“false”），将配置项“spark.inputFormat.cache.enabled”设置为“false”。

数据规划 在客户端执行hbase shell进入HBase命令行。 在hbase命令执行下面的命令创建HBbase表： create 'streamingTable','cf1' 在客户端另外一个session通过linux命令构造一个端口进行接收数据（不同操作系统的机器，命令可能不同，suse尝试使用netcat -lk 9999）： nc -lk 9999 提交任务命令执行之后，在该命令下输入要提交的数据，通过HBase表进行接收。 在构造一个端口进行接收数据时，需要在客户端所在服务器上安装netcat。

打包项目 通过IDEA自带的Maven工具，打包项目，生成jar包。具体操作请参考在Linux环境中调测Spark应用。 将打包生成的jar包上传到Spark客户端所在服务器的任意目录（例如“$SPARK_HOME” ）下。 将user.keytab、krb5.conf 两个文件上传客户端所在服务器上（文件上传的路径需要和生成的jar包路径一致）。 若运行“Spark on HBase”样例程序，需要在Spark客户端的“spark-defaults.conf”配置文件中将配置项“spark.yarn.security.credentials.hbase.enabled”设置为“true”（该参数值默认为“false”，改为“true”后对已有业务没有影响。如果要卸载HBase服务，卸载前请将此参数值改回“false”），将配置项“spark.inputFormat.cache.enabled”设置为“false”。

提交命令 假设用例代码打包后的jar包名为spark-hbaseContext-test-1.0.jar，并将jar包放在客户端“$SPARK_HOME”目录下，以下命令均在“$SPARK_HOME”目录执行，Java接口对应的类名前有Java字样，请参考具体样例代码进行书写。 yarn-client模式： java/scala版本（类名等请与实际代码保持一致，此处仅为示例） bin/spark-submit --master yarn --deploy-mode client --class com.huawei.bigdata.spark.examples.hbasecontext.JavaHBaseForEachPartitionExample SparkOnHbaseJavaExample.jar table2 cf1 python版本（文件名等请与实际保持一致，此处仅为示例），假设对应的Java代码打包后包名为SparkOnHbaseJavaExample.jar，且放在当前提交目录。 bin/spark-submit --master yarn --deploy-mode client --jars SparkOnHbaseJavaExample.jar HBaseForEachPartitionExample.py table2 cf1 yarn-cluster模式： java/scala版本（类名等请与实际代码保持一致，此处仅为示例） bin/spark-submit --master yarn --deploy-mode cluster --class com.huawei.bigdata.spark.examples.hbasecontext.JavaHBaseForEachPartitionExample --files /opt/user.keytab,/opt/krb5.conf SparkOnHbaseJavaExample.jar table2 cf1 python版本（文件名等请与实际保持一致，此处仅为示例），假设对应的Java代码打包后包名为SparkOnHbaseJavaExample.jar，且放在当前提交目录。 bin/spark-submit --master yarn --deploy-mode cluster --files /opt/user.keytab,/opt/krb5.conf --jars SparkOnHbaseJavaExample.jar HBaseForEachPartitionExample.py table2 cf1

Python样例代码 下面代码片段仅为演示，具体代码参见SparkOnHbasePythonExample中HBaseStreamingBulkPutExample文件： # -*- coding:utf-8 -*-"""【说明】(1)由于pyspark不提供Hbase相关api,本样例使用Python调用Java的方式实现(2)如果使用yarn-client模式运行,请确认Spark2x客户端Spark2x/spark/conf/spark-defaults.conf中 spark.yarn.security.credentials.hbase.enabled参数配置为true"""from py4j.java_gateway import java_importfrom pyspark.sql import SparkSession# 创建SparkSessionspark = SparkSession\ .builder\ .appName("JavaHBaseStreamingBulkPutExample")\ .getOrCreate()# 向sc._jvm中导入要运行的类java_import(spark._jvm, 'com.huawei.bigdata.spark.examples.streaming.JavaHBaseStreamingBulkPutExample')# 创建类实例并调用方法，传递sc._jsc参数spark._jvm.JavaHBaseStreamingBulkPutExample().execute(spark._jsc, sys.argv)# 停止SparkSessionspark.stop()

Python样例代码 下面代码片段仅为演示，具体代码参见SparkOnHbasePythonExample中HBaseForEachPartitionExample文件： # -*- coding:utf-8 -*-"""【说明】(1)由于pyspark不提供Hbase相关api,本样例使用Python调用Java的方式实现(2)如果使用yarn-client模式运行,请确认Spark2x客户端Spark2x/spark/conf/spark-defaults.conf中 spark.yarn.security.credentials.hbase.enabled参数配置为true"""from py4j.java_gateway import java_importfrom pyspark.sql import SparkSession# 创建SparkSessionspark = SparkSession\ .builder\ .appName("JavaHBaseForEachPartitionExample")\ .getOrCreate()# 向sc._jvm中导入要运行的类java_import(spark._jvm, 'com.huawei.bigdata.spark.examples.hbasecontext.JavaHBaseForEachPartitionExample')# 创建类实例并调用方法，传递sc._jsc参数spark._jvm.JavaHBaseForEachPartitionExample().execute(spark._jsc, sys.argv)# 停止SparkSessionspark.stop()

提交命令 假设用例代码打包后的jar包名为spark-hbaseContext-test-1.0.jar，并将jar包放在客户端“$SPARK_HOME”目录下，以下命令均在“$SPARK_HOME”目录执行，Java接口对应的类名前有Java字样，请参考具体样例代码进行书写。 yarn-client模式： java/scala版本（类名等请与实际代码保持一致，此处仅为示例），${ip}请使用实际执行nc -lk 9999的命令的机器ip bin/spark-submit --master yarn --deploy-mode client --class com.huawei.bigdata.spark.examples.streaming.JavaHBaseStreamingBulkPutExample SparkOnHbaseJavaExample.jar ${ip} 9999 streamingTable cf1 python版本（文件名等请与实际保持一致，此处仅为示例），假设对应的Java代码打包后包名为SparkOnHbaseJavaExample.jar，且放在当前提交目录。 bin/spark-submit --master yarn --deploy-mode client --jars SparkOnHbaseJavaExample.jar HBaseStreamingBulkPutExample.py ${ip} 9999 streamingTable cf1 yarn-cluster模式： java/scala版本（类名等请与实际代码保持一致，此处仅为示例），${ip}请使用实际执行nc -lk 9999的命令的机器ip bin/spark-submit --master yarn --deploy-mode cluster --class com.huawei.bigdata.spark.examples.streaming.JavaHBaseStreamingBulkPutExample --files /opt/user.keytab,/opt/krb5.conf SparkOnHbaseJavaExample.jar ${ip} 9999 streamingTable cf1 python版本（文件名等请与实际保持一致，此处仅为示例），假设对应的Java代码打包后包名为SparkOnHbaseJavaExample.jar，且放在当前提交目录。 bin/spark-submit --master yarn --deploy-mode cluster --files /opt/user.keytab,/opt/krb5.conf --jars SparkOnHbaseJavaExample.jar HBaseStreamingBulkPutExample.py ${ip} 9999 streamingTable cf1

打包项目 将user.keytab、krb5.conf 两个文件上传客户端所在服务器上。 通过IDEA自带的Maven工具，打包项目，生成jar包。具体操作请参考在Linux环境中调测Spark应用。 编译打包前，样例代码中的user.keytab、krb5.conf文件路径需要修改为该文件所在客户端服务器的实际路径。例如：“/opt/female/user.keytab”，“/opt/female/krb5.conf”。 运行Python样例代码无需通过Maven打包，只需要上传user.keytab、krb5.conf 文件到客户端所在服务器上。 将打包生成的jar包上传到Spark客户端所在服务器的任意目录（例如“ /opt/female/” ）下。

提交命令 假设用例代码打包后的jar包名为spark-hbaseContext-test-1.0.jar，并将jar包放在客户端“$SPARK_HOME”目录下，以下命令均在“$SPARK_HOME”目录执行。 yarn-client模式： java/scala版本（类名等请与实际代码保持一致，此处仅为示例） bin/spark-submit --master yarn --deploy-mode client --jars /opt/female/protobuf-java-2.5.0.jar --conf spark.yarn.user.classpath.first=true --class com.huawei.bigdata.spark.examples.datasources.HBaseSource SparkOnHbaseJavaExample.jar python版本（文件名等请与实际保持一致，此处仅为示例），假设对应的Java代码打包后包名为SparkOnHbaseJavaExample.jar，且放在当前提交目录。 bin/spark-submit --master yarn --deploy-mode client --conf spark.yarn.user.classpath.first=true --jars SparkOnHbaseJavaExample.jar,/opt/female/protobuf-java-2.5.0.jar HBaseSource.py yarn-cluster模式： java/scala版本（类名等请与实际代码保持一致，此处仅为示例） bin/spark-submit --master yarn --deploy-mode cluster --jars /opt/female/protobuf-java-2.5.0.jar --conf spark.yarn.user.classpath.first=true --class com.huawei.bigdata.spark.examples.datasources.HBaseSource --files /opt/user.keytab,/opt/krb5.conf SparkOnHbaseJavaExample.jar python版本（文件名等请与实际保持一致，此处仅为示例），假设对应的Java代码打包后包名为SparkOnHbaseJavaExample.jar，且放在当前提交目录。 bin/spark-submit --master yarn --files /opt/user.keytab,/opt/krb5.conf --conf spark.yarn.user.classpath.first=true --jars SparkOnHbaseJavaExample.jar,/opt/female/protobuf-java-2.5.0.jar HBaseSource.py

打包项目 通过IDEA自带的Maven工具，打包项目，生成jar包。具体操作请参考在Linux环境中调测Spark应用。 将打包生成的jar包上传到Spark客户端所在服务器的任意目录（例如“$SPARK_HOME” ）下。 将user.keytab、krb5.conf 两个文件上传客户端所在服务器上（文件上传的路径需要和生成的jar包路径一致）。 若运行“Spark on HBase”样例程序，需要在Spark客户端的“spark-defaults.conf”配置文件中将配置项“spark.yarn.security.credentials.hbase.enabled”设置为“true”（该参数值默认为“false”，改为“true”后对已有业务没有影响。如果要卸载HBase服务，卸载前请将此参数值改回“false”），将配置项“spark.inputFormat.cache.enabled”设置为“false”。

Python样例代码 下面代码片段仅为演示，具体代码参见SparkOnHbasePythonExample中的HBaseSource文件： # -*- coding:utf-8 -*-"""【说明】(1)由于pyspark不提供Hbase相关api,本样例使用Python调用Java的方式实现(2)如果使用yarn-client模式运行,请确认Spark2x客户端Spark2x/spark/conf/spark-defaults.conf中 spark.yarn.security.credentials.hbase.enabled参数配置为true"""from py4j.java_gateway import java_importfrom pyspark.sql import SparkSession# 创建SparkSessionspark = SparkSession\ .builder\ .appName("HBaseSourceExample")\ .getOrCreate()# 向sc._jvm中导入要运行的类java_import(spark._jvm, 'com.huawei.bigdata.spark.examples.datasources.HBaseSource')# 创建类实例并调用方法，传递sc._jsc参数spark._jvm.HBaseSource().execute(spark._jsc)# 停止SparkSessionspark.stop()

场景说明 假定用户有某个周末网民网购停留时间的日志文本，基于某些业务要求，要求开发Spark应用程序实现如下功能： 统计日志文件中本周末网购停留总时间超过2个小时的女性网民信息。 周末两天的日志文件第一列为姓名，第二列为性别，第三列为本次停留时间，单位为分钟，分隔符为“,”。 log1.txt：周六网民停留日志 LiuYang,female,20YuanJing,male,10GuoYijun,male,5CaiXuyu,female,50Liyuan,male,20FangBo,female,50LiuYang,female,20YuanJing,male,10GuoYijun,male,50CaiXuyu,female,50FangBo,female,60 log2.txt：周日网民停留日志 LiuYang,female,20YuanJing,male,10CaiXuyu,female,50FangBo,female,50GuoYijun,male,5CaiXuyu,female,50Liyuan,male,20CaiXuyu,female,50FangBo,female,50LiuYang,female,20YuanJing,male,10FangBo,female,50GuoYijun,male,50CaiXuyu,female,50FangBo,female,60

常用接口 YARN常用的Java类有如下几个。 ApplicationClientProtocol 用于Client与ResourceManager之间。Client通过该协议可实现将应用程序提交到ResourceManager上，查询应用程序的运行状态或者中止应用程序等功能。 表1 ApplicationClientProtocol常用方法 方法 说明 forceKillApplication(KillApplicationRequest request) Client通过此接口请求RM中止一个已提交的任务。 getApplicationAttemptReport(GetApplicationAttemptReportRequest request) Client通过此接口从RM获取指定ApplicationAttempt的报告信息。 getApplicationAttempts(GetApplicationAttemptsRequest request) Client通过此接口从RM获取所有ApplicationAttempt的报告信息。 getApplicationReport(GetApplicationReportRequest request) Client通过此接口从RM获取某个应用的报告信息。 getApplications(GetApplicationsRequest request) Client通过此接口从RM获取满足一定过滤条件的应用的报告信息。 getClusterMetrics(GetClusterMetricsRequest request) Client通过此接口从RM获取集群的Metrics。 getClusterNodes(GetClusterNodesRequest request) Client通过此接口从RM获取集群中的所有节点信息。 getContainerReport(GetContainerReportRequest request) Client通过此接口从RM获取某个Container的报告信息。 getContainers(GetContainersRequest request) Client通过此接口从RM获取某个ApplicationAttemp的所有Container的报告信息。 getDelegationToken(GetDelegationTokenRequest request) Client通过此接口获取授权票据，用于container访问相应的service。 getNewApplication(GetNewApplicationRequest request) Client通过此接口获取一个新的应用ID号，用于提交新的应用。 getQueueInfo(GetQueueInfoRequest request) Client通过此接口从RM中获取队列的相关信息。 getQueueUserAcls(GetQueueUserAclsInfoRequest request) Client通过此接口从RM中获取当前用户的队列访问权限信息。 moveApplicationAcrossQueues(MoveApplicationAcrossQueuesRequest request) 移动一个应用到新的队列。 submitApplication(SubmitApplicationRequest request) Client通过此接口提交一个新的应用到RM。 ApplicationMasterProtocol 用于ApplicationMaster与ResourceManager之间。ApplicationMaster使用该协议向ResourceManager注册、申请资源、获取各个任务的运行情况等。 表2 ApplicationMasterProtocol常用方法 方法 说明 allocate(AllocateRequest request) AM通过此接口提交资源分配申请。 finishApplicationMaster(FinishApplicationMasterRequest request) AM通过此接口通知RM其运行成功或者失败。 registerApplicationMaster(RegisterApplicationMasterRequest request) AM通过此接口向RM进行注册。 ContainerManagementProtocol 用于ApplicationMaster与NodeManager之间。ApplicationMaster使用该协议要求NodeManager启动/中止Container或者查询Container的运行状态。 表3 ContainerManagementProtocol常用方法 方法 说明 getContainerStatuses(GetContainerStatusesRequest request) AM通过此接口向NM请求Containers的当前状态信息。 startContainers(StartContainersRequest request) AM通过此接口向NM提供需要启动的containers列表的请求。 stopContainers(StopContainersRequest request) AM通过此接口请求NM停止一系列已分配的Containers。

SparkSQL常用接口 Spark SQL中常用的类有： SQLContext：是Spark SQL功能和DataFrame的主入口。 DataFrame：是一个以命名列方式组织的分布式数据集。 HiveContext：获取存储在Hive中数据的主入口。 表6 常用的Actions方法 方法 说明 collect(): Array[Row] 返回一个数组，包含DataFrame的所有列。 count(): Long 返回DataFrame中的行数。 describe(cols: String*): DataFrame 计算统计信息，包含计数，平均值，标准差，最小值和最大值。 first(): Row 返回第一行。 Head(n:Int): Row 返回前n行。 show(numRows: Int, truncate: Boolean): Unit 用表格形式显示DataFrame。 take(n:Int): Array[Row] 返回DataFrame中的前n行。 表7 基本的DataFrame Functions 方法 说明 explain(): Unit 打印出SQL语句的逻辑计划和物理计划。 printSchema(): Unit 打印schema信息到控制台。 registerTempTable(tableName: String): Unit 将DataFrame注册为一张临时表，其周期和SQLContext绑定在一起。 toDF(colNames: String*): DataFrame 返回一个列重命名的DataFrame。

原因分析 以自定义UDF为例： 报错信息显示是找不到类。 首先需要确认的是这个类属于的jar包是否在jvm的classpath里面， spark自带的jar都在“spark客户端目录/jars/”。 确认是否存在多个jar包拥有这个类。 如果是其他依赖包，可能是没有使用--jars添加到任务里面。 如果是已经添加到任务里面，但是依旧没有取到，可能是因为配置文件的driver或者executor的classpath配置不正确，可以查看日志确认是否加载到环境。 另外可能报错是类初始化失败导致后面使用这个类的时候出现上述报错，需要确认是否在之前就有初始化失败或者其他报错的情况发生。 报错信息显示找不到方法。 确认这个方法对应的类所在的jar包是否加载到jvm的classpath里面，spark自带的类都在“spark客户端目录/jars/”。 确认是否有多个jar包包含这个类（尤其注意相同工具的不同版本）。 如果报错是Hadoop相关的包，有可能是因为使用的Hadoop版本不一致导致部分方法已经更改。 如果报错的是三方包里面的类，可能是因为Spark已经自带了相关的jar包，但是和代码中使用的版本不一致。

问题现象 Spark能对接很多的第三方工具，因此在使用过程中经常会依赖一堆的三方包。而有一些包 MRS 已经自带，这样就有可能造成代码使用的jar包版本和集群自带的jar包版本不一致，在使用过程中就有可能出现jar包冲突的情况。 常见的jar包冲突报错有： 1、报错类找不到：java.lang.NoClassDefFoundError 2、报错方法找不到：java.lang.NoSuchMethodError

Spark Streaming常用接口 Spark Streaming中常见的类有： StreamingContext：是Spark Streaming功能的主入口，负责提供创建DStreams的方法，入参中需要设置批次的时间间隔。 dstream.DStream：是一种代表RDDs连续序列的数据类型，代表连续数据流。 dstream.PariDStreamFunctions：键值对的DStream，常见的操作如groupByKey和reduceByKey。 对应的Spark Streaming的JAVA API是JavaStreamingContext，JavaDStream和JavaPairDStream。 Spark Streaming的常见方法与Spark Core类似，下表罗列了Spark Streaming特有的一些方法。 表4 Spark Streaming方法介绍 方法 说明 socketTextStream(hostname: String, port: Int, storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2): ReceiverInputDStream[String] 从TCP源主机：端口创建一个输入流。 start():Unit 启动Spark Streaming计算。 awaitTermination(timeout: long):Unit 当前进程等待终止，如Ctrl+C等。 stop(stopSparkContext: Boolean, stopGracefully: Boolean): Unit 终止Spark Streaming计算。 transform[T](dstreams: Seq[DStream[_]], transformFunc: (Seq[RDD[_]], Time) ? RDD[T])(implicit arg0: ClassTag[T]): DStream[T] 对每一个RDD应用function操作得到一个新的DStream。 UpdateStateByKey(func) 更新DStream的状态。使用此方法，需要定义状态和状态更新函数。 window(windowLength, slideInterval) 根据源DStream的窗口批次计算得到一个新的DStream。 countByWindow(windowLength, slideInterval) 返回流中滑动窗口元素的个数。 reduceByWindow(func, windowLength, slideInterval) 当调用在DStream的KV对上，返回一个新的DStream的KV对，其中每个Key的Value根据滑动窗口中批次的reduce函数聚合得到。 join(otherStream, [numTasks]) 实现不同的Spark Streaming之间做合并操作。 DStreamKafkaWriter.writeToKafka() 支持将DStream中的数据批量写入到Kafka。 DStreamKafkaWriter.writeToKafkaBySingle() 支持将DStream中的数据逐条写入到Kafka。 表5 Spark Streaming增强特性接口 方法 说明 DStreamKafkaWriter.writeToKafka() 支持将DStream中的数据批量写入到Kafka。 DStreamKafkaWriter.writeToKafkaBySingle() 支持将DStream中的数据逐条写入到Kafka。