华为云用户手册

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()

打包项目 通过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”。

代码样例 由于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()

提交命令 假设用例代码打包后的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

场景说明 假定用户有某个周末网民网购停留时间的日志文本，基于某些业务要求，要求开发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

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()

打包项目 将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”。

常用接口 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。

问题现象 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。

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

简介 JD BCS erver是Hive中的HiveServer2的另外一个实现，它底层使用了Spark SQL来处理SQL语句，从而比Hive拥有更高的性能。 JDB CS erver是一个JDBC接口，用户可以通过JDBC连接JDBCServer来访问SparkSQL的数据。JDBCServer在启动的时候，会启动一个sparkSQL的应用程序，而通过JDBC连接进来的客户端共同分享这个sparkSQL应用程序的资源，也就是说不同的用户之间可以共享数据。JDBCServer启动时还会开启一个侦听器，等待JDBC客户端的连接和提交查询。所以，在配置JDBCServer的时候，至少要配置JDBCServer的主机名和端口，如果要使用hive数据的话，还要提供hive metastore的uris。 JDBCServer默认在安装节点上的22550端口起一个JDBC服务（通过参数hive.server2.thrift.port配置），可以通过Beeline或者JDBC客户端代码来连接它，从而执行SQL命令。 如果您需要了解JDBCServer的其他信息，请参见Spark官网：http://archive.apache.org/dist/spark/docs/3.3.1/sql-programming-guide.html#distributed-sql-engine。

增强特性 对比开源社区，华为还提供了两个增强特性，JDBCServerHA方案和设置JDBCServer连接的超时时间。 JDBCServer的HA方案，多个JDBCServer主节点同时提供服务，当其中一个节点发生故障时，新的客户端连接会分配到其他主节点上，从而保障无间断为集群提供服务。Beeline和JDBC客户端代码两种连接方式的操作相同。 设置客户端与JDBCServer连接的超时时间。 Beeline 在网络拥塞的情况下，这个特性可以避免beeline由于无限等待服务端的返回而挂起。使用方式如下： 启动beeline时，在后面追加“--socketTimeOut=n”，其中“n”表示等待服务返回的超时时长，单位为秒，默认为“0”（表示永不超时）。建议根据业务场景，设置为业务所能容忍的最大等待时长。 JDBC客户端代码 在网络拥塞的情况下，这个特性可以避免客户端由于无限等待服务端的返回而挂起。使用方式如下： 在执行“DriverManager.getConnection”方法获取JDBC连接前，添加“DriverManager.setLoginTimeout(n)”方法来设置超时时长，其中n表示等待服务返回的超时时长，单位为秒，类型为Int，默认为“0”（表示永不超时）。建议根据业务场景，设置为业务所能容忍的最大等待时长。

数据规划 StructuredStreaming样例工程的数据存储在Kafka组件中。向Kafka组件发送数据（需要有Kafka权限用户）。 确保集群安装完成，包括HDFS、Yarn、Spark和Kafka。 将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 1 --topic {Topic} 启动Kafka的Producer，向Kafka发送数据。 {ClassPath}表示工程jar包的存放路径，详细路径由用户指定，可参考在Linux环境中调测Spark应用章节中导出jar包的操作步骤。 java -cp $SPARK_HOME/jars/*:$SPARK_HOME/jars/streamingClient010/*:{ClassPath} com.huawei.bigdata.spark.examples.KafkaWordCountProducer {BrokerList} {Topic} {messagesPerSec} {wordsPerMessage}

打包项目 将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目录下获取）。

打包项目 通过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中HBaseBulkLoadPythonExample文件： # -*- 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("JavaHBaseBulkLoadExample")\ .getOrCreate()# 向sc._jvm中导入要运行的类java_import(spark._jvm, 'com.huawei.bigdata.spark.examples.HBaseBulkLoadPythonExample')# 创建类实例并调用方法，传递sc._jsc参数spark._jvm.HBaseBulkLoadPythonExample().hbaseBulkLoad(spark._jsc, sys.argv[1], sys.argv[2])# 停止SparkSessionspark.stop()

数据规划 使用Spark-Beeline工具创建Spark和HBase表table1、table2，并通过HBase插入数据。 确保JDBCServer已启动。登录Spark2x客户端节点。 使用Spark-beeline工具创建Spark表table1。 create table table1 ( key string, cid string ) using org.apache.spark.sql.hbase.HBaseSource options( hbaseTableName "table1", keyCols "key", colsMapping "cid=cf.cid"); 通过HBase插入数据，命令如下： put 'table1', '1', 'cf:cid', '100' 使用Spark-Beeline工具创建Spark表table2。 create table table2 ( key string, cid string ) using org.apache.spark.sql.hbase.HBaseSource options( hbaseTableName "table2", keyCols "key", colsMapping "cid=cf.cid"); 通过HBase插入数据，命令如下： put 'table2', '1', 'cf:cid', '1000'

运行任务 进入Spark客户端目录，调用bin/spark-submit脚本运行代码，运行命令分别如下（类名与文件名等请与实际代码保持一致，此处仅为示例）： 运行Java或Scala样例代码 bin/spark-submit --jars {客户端安装路径}/Spark/spark/jars/protobuf-java-2.5.0.jar --conf spark.yarn.user.classpath.first=true --class com.huawei.bigdata.spark.examples.SparkHbasetoHbase --master yarn --deploy-mode client /opt/female/SparkHbasetoHbase-1.0.jar 运行Python样例程序 由于pyspark不提供Hbase相关api，本样例使用Python调用Java的方式实现。将所提供 Java代码使用maven打包成jar，并放在相同目录下，运行python程序时要使用--jars把jar包加载到classpath中。 由于Python样例代码中未给出认证信息，请在执行应用程序时通过配置项“--keytab”和“--principal”指定认证信息。 bin/spark-submit --master yarn --deploy-mode client --keytab /opt/FIclient/user.keytab --principal sparkuser --conf spark.yarn.user.classpath.first=true --jars /opt/female/SparkHbasetoHbasePythonExample/SparkHbasetoHbase-1.0.jar,/opt/female/protobuf-java-2.5.0.jar /opt/female/SparkHbasetoHbasePythonExample/SparkHbasetoHbasePythonExample.py

提交命令 假设用例代码打包后的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.JavaHBaseBulkLoadExample SparkOnHbaseJavaExample.jar /tmp/hfile bulkload-table-test python版本（文件名等请与实际保持一致，此处仅为示例），假设对应的Java代码打包后包名为SparkOnHbaseJavaExample.jar，且放在当前提交目录。 bin/spark-submit --master yarn --deploy-mode client --jars SparkOnHbaseJavaExample.jar HBaseBulkLoadExample.py /tmp/hfile bulkload-table-test yarn-cluster模式： java/scala版本（类名等请与实际代码保持一致，此处仅为示例） bin/spark-submit --master yarn --deploy-mode cluster --class com.huawei.bigdata.spark.examples.hbasecontext.JavaHBaseBulkLoadExample --files /opt/user.keytab,/opt/krb5.conf SparkOnHbaseJavaExample.jar /tmp/hfile bulkload-table-test python版本（文件名等请与实际保持一致，此处仅为示例），假设对应的Java代码打包后包名为SparkOnHbaseJavaExample.jar，且放在当前提交目录。 bin/spark-submit --master yarn --deploy-mode cluster --files /opt/user.keytab,/opt/krb5.conf --jars SparkOnHbaseJavaExample.jar HBaseBulkLoadExample.py /tmp/hfile bulkload-table-test

场景说明 假定HBase的table1表存储用户当天消费的金额信息，table2表存储用户历史消费的金额信息。 现table1表有记录key=1,cf:cid=100，表示用户1在当天消费金额为100元。 table2表有记录key=1,cf:cid=1000，表示用户1的历史消息记录金额为1000元。 基于某些业务要求，要求开发Spark应用程序实现如下功能： 根据用户名累计用户的历史消费金额，即用户总消费金额=100(用户当天的消费金额) + 1000(用户历史消费金额)。 上例所示，运行结果table2表用户key=1的总消费金额为cf:cid=1100元。

场景说明 用户可以在Spark应用程序中使用HBaseContext的方式去使用HBase，将要插入的数据的rowKey构造成rdd，然后通过HBaseContext的bulkLoad接口将rdd写入HFile中。将生成的HFile文件导入HBase表的操作采用如下格式的命令，不属于本接口范围，不在此进行详细说明： hbase org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles {hfilePath} {tableName}

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

功能简介 Spark的REST API以JSON格式展现Web UI的一些指标，提供用户一种更简单的方法去创建新的展示和监控的工具，并且支持查询正在运行的app和已经结束的app的相关信息。开源的Spark REST接口支持对Jobs、Stages、Storage、Environment和Executors的信息进行查询， FusionInsight 版本中添加了查询SQL、JDBC Server和Streaming的信息的REST接口。开源REST接口完整和详细的描述请参考官网上的文档以了解其使用方法：https://archive.apache.org/dist/spark/docs/3.3.1/monitoring.html#rest-api。

REST API增强 SQL相关的命令：获取所有SQL语句和执行时间最长的SQL语句 SparkUI命令： curl -k -i --negotiate -u: "https://192.168.195.232:8090/proxy/application_1476947670799_0053/api/v1/applications/application_1476947670799_0053/SQL" 其中192.168.195.232为ResourceManager主节点的业务IP，8090为ResourceManager的端口号，application_1476947670799_0053为在YARN中的应用ID。 可以在命令后的url路径增加相应的参数设置，搜索对应的SQL语句。 例如，查看100条sql语句： curl -k -i --negotiate -u: "https://192.168.195.232:8090/proxy/application_1476947670799_0053/api/v1/applications/application_1476947670799_0053/SQL?limit=100" 查看正在运行的参数： curl -k -i --negotiate -u: "https://192.168.195.232:8090/proxy/application_1476947670799_0053/api/v1/applications/application_1476947670799_0053/SQL?completed=false" JobHistory命令： curl -k -i --negotiate -u: "https://192.168.227.16:4040/api/v1/applications/application_1478570725074_0004/SQL" 其中192.168.227.16为JobHistory节点的业务IP，4040为JobHistory的端口号，application_1478570725074_0004为应用ID。 结果： SparkUI命令和JobHistory命令的查询结果均为： { "longestDurationOfCompletedSQL" : [ { "id" : 0, "status" : "COMPLETED", "description" : "getCallSite at SQLExecution.scala:48", "submissionTime" : "2016/11/08 15:39:00", "duration" : "2 s", "runningJobs" : [ ], "successedJobs" : [ 0 ], "failedJobs" : [ ] } ], "sqls" : [ { "id" : 0, "status" : "COMPLETED", "description" : "getCallSite at SQLExecution.scala:48", "submissionTime" : "2016/11/08 15:39:00", "duration" : "2 s", "runningJobs" : [ ], "successedJobs" : [ 0 ], "failedJobs" : [ ] }]} 结果分析： 通过这个命令，可以查询当前应用的所有SQL语句的信息（即结果中“sqls”的部分），执行时间最长的SQL语句的信息（即结果中“longestDurationOfCompletedSQL”的部分）。每个SQL语句的信息如下表3。 表3 SQL的常用信息 参数 描述 id SQL语句的ID status SQL语句的执行状态，有RUNNING、COMPLETED、FAILED三种 runningJobs SQL语句产生的job中，正在执行的job列表 successedJobs SQL语句产生的job中，执行成功的job列表 failedJobs SQL语句产生的job中，执行失败的job列表 JDBC Server相关的命令：获取连接数，正在执行的SQL数，所有session信息，所有SQL的信息 命令： curl -k -i --negotiate -u: "https://192.168.195.232:8090/proxy/application_1476947670799_0053/api/v1/applications/application_1476947670799_0053/sqlserver" 其中192.168.195.232为ResourceManager主节点的业务IP，8090为ResourceManager的端口号，application_1476947670799_0053为在YARN中的应用ID。 结果： { "sessionNum" : 1, "runningSqlNum" : 0, "sessions" : [ { "user" : "spark", "ip" : "192.168.169.84", "sessionId" : "9dfec575-48b4-4187-876a-71711d3d7a97", "startTime" : "2016/10/29 15:21:10", "finishTime" : "", "duration" : "1 minute 50 seconds", "totalExecute" : 1 } ], "sqls" : [ { "user" : "spark", "jobId" : [ ], "groupId" : "e49ff81a-230f-4892-a209-a48abea2d969", "startTime" : "2016/10/29 15:21:13", "finishTime" : "2016/10/29 15:21:14", "duration" : "555 ms", "statement" : "show tables", "state" : "FINISHED", "detail" : "== Parsed Logical Plan ==\nShowTablesCommand None\n\n== Analyzed Logical Plan ==\ntableName: string, isTemporary: boolean\nShowTablesCommand None\n\n== Cached Logical Plan ==\nShowTablesCommand None\n\n== Optimized Logical Plan ==\nShowTablesCommand None\n\n== Physical Plan ==\nExecutedCommand ShowTablesCommand None\n\nCode Generation: true" } ]} 结果分析： 通过这个命令，可以查询当前JDBC应用的session连接数，正在执行的SQL数，所有的session和SQL信息。每个session的信息如下表4，每个SQL的信息如下表5。 表4 session常用信息 参数 描述 user 该session连接的用户 ip session所在的节点IP sessionId session的ID startTime session开始连接的时间 finishTime session结束连接的时间 duration session连接时长 totalExecute 在该session上执行的SQL数 表5 sql常用信息 参数 描述 user SQL执行的用户 jobId SQL语句包含的job id列表 groupId SQL所在的group id startTime SQL开始时间 finishTime SQL结束时间 duration SQL执行时长 statement 对应的语句 detail 对应的逻辑计划，物理计划 JDBC api增强通过beeline里面获取的executionID 取消当前正在执行的SQL 命令： curl -k -i --negotiate -X PUT -u: "https://192.168.195.232:8090/proxy/application_1477722033672_0008/api/v1/applications/application_1477722033672_0008/cancel/execution?executionId=8" 结果： 取消executionId 执行序号为8的job任务。 补充说明： spark-beeline里面执行SQL语句，如果该SQL语句产生spark任务，该SQL的executionId将会被打印在beeline里面，这个时候如果想取消这条sql的执行，可以用上述命令。 Streaming相关的命令：获取平均输入频率，平均调度时延，平均执行时长，总时延平均值 命令： curl -k -i --negotiate -u: "https://192.168.195.232:8090/proxy/application_1477722033672_0008/api/v1/applications/application_1477722033672_0008/streaming/statistics" 其中192.168.195.232为ResourceManager主节点的业务IP，8090为ResourceManager的端口号，application_1477722033672_0008为在YARN中的应用ID。 结果： {"startTime" : "2018-12-25T08:58:10.836GMT", "batchDuration" : 1000, "numReceivers" : 1, "numActiveReceivers" : 1, "numInactiveReceivers" : 0, "numTotalCompletedBatches" : 373, "numRetainedCompletedBatches" : 373, "numActiveBatches" : 0, "numProcessedRecords" : 1, "numReceivedRecords" : 1, "avgInputRate" : 0.002680965147453083, "avgSchedulingDelay" : 14, "avgProcessingTime" : 47, "avgTotalDelay" : 62} 结果分析： 通过这个命令，可以查询当前Streaming应用的平均输入频率（events/sec），平均调度时延（ms），平均执行时长（ms），总时延平均值（ms）。

回答 由于checkpoint中包含了spark应用的对象序列化信息、task执行状态信息、配置信息等，因此，当存在以下问题时，从checkpoint恢复spark应用将会失败。 业务代码变更且变更类未明确指定SerialVersionUID。 spark内部类变更，且变更类未明确指定SerialVersionUID。 另外，由于checkpoint保存了部分配置项，因此可能导致业务修改了部分配置项后，从checkpoint恢复时，配置项依然保持为旧值的情况。当前只有以下部分配置会在从checkpoint恢复时重新加载。 "spark.yarn.app.id", "spark.yarn.app.attemptId", "spark.driver.host", "spark.driver.bindAddress", "spark.driver.port", "spark.master", "spark.yarn.jars", "spark.yarn.keytab", "spark.yarn.principal", "spark.yarn.credentials.file", "spark.yarn.credentials.renewalTime", "spark.yarn.credentials.updateTime", "spark.ui.filters", "spark.mesos.driver.frameworkId", "spark.yarn.jars"

回答 问题原因： 在IBM JDK下建立的JDBC connection时间超过登录用户的认证超时时间（默认一天），导致认证失败。 IBM JDK的机制跟Oracle JDK的机制不同，IBM JDK在认证登录后的使用过程中做了时间检查却没有检测外部的时间更新，导致即使显式调用relogin也无法得到刷新。 解决措施： 通常情况下，在发现JDBC connection不可用的时候，可以关闭该connection，重新创建一个connection继续执行。