华为云用户手册

Kafka应用开发流程介绍 Kafka客户端角色包括Producer和Consumer两个角色，其应用开发流程是相同的。 开发流程中各个阶段的说明如图1和表1所示。 图1 Kafka客户端程序开发流程 表1 Kafka客户端开发的流程说明 阶段 说明 参考文档 准备开发环境 Kafka的客户端程序当前推荐使用java语言进行开发，可使用IntelliJ IDEA工具开发。 Kafka的运行环境即Kafka客户端，请根据指导完成客户端的安装和配置。 准备本地应用开发环境 准备连接集群配置文件 应用程序开发或运行过程中，需通过集群相关配置文件信息连接 MRS 集群，配置文件通常包括集群组件信息文件以及用于安全认证的用户文件，可从已创建好的MRS集群中获取相关内容。 用于程序调测或运行的节点，需要与MRS集群内节点网络互通，同时配置hosts 域名 信息。 准备连接Kafka集群配置文件 配置并导入样例工程 Kafka提供了不同场景下的样例程序，您可以导入样例工程进行程序学习。 导入并配置Kafka样例工程 根据业务场景开发工程 提供了Producer和Consumer相关API的使用样例，包含了API和多线程的使用场景，帮助用户快速熟悉Kafka接口。 开发Kafka应用 编译与运行程序 指导用户将开发好的程序编译并提交运行。 调测Kafka应用 父主题： Kafka开发指南（普通模式）

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

Kafka开发应用时，需要准备的开发和运行环境如表1所示： 表1 开发环境 准备项 说明 操作系统 开发环境：Windows系统，支持Windows 7以上版本。 运行环境：Windows系统或Linux系统。 如需在本地调测程序，运行环境需要和集群业务平面网络互通。 安装和配置IntelliJ IDEA 开发环境的基本配置。版本要求：JDK使用1.8版本，IntelliJ IDEA使用2019.1或其他兼容版本。 说明： 若使用IBM JDK，请确保IntelliJ IDEA中的JDK配置为IBM JDK。 若使用Oracle JDK，请确保IntelliJ IDEA中的JDK配置为Oracle JDK。 若使用Open JDK，请确保IntelliJ IDEA中的JDK配置为Open JDK。 安装JDK 开发和运行环境的基本配置。版本要求如下： 服务端和客户端仅支持自带的OpenJDK，版本为1.8.0_272，不允许替换。 对于客户应用需引用SDK类的Jar包运行在客户应用进程中的。 X86客户端： Oracle JDK：支持1.8版本 IBM JDK：支持1.8.5.11版本 TaiShan客户端： OpenJDK：支持1.8.0_272版本 说明： 基于安全考虑，服务端只支持TLS V1.2及以上的加密协议。 IBM JDK默认只支持TLS V1.0，若使用IBM JDK，请配置启动参数“com.ibm.jsse2.overrideDefaultTLS”为“true”，设置后可以同时支持TLS V1.0/V1.1/V1.2，详情参见https://www.ibm.com/support/knowledgecenter/zh/SSYKE2_8.0.0/com.ibm.java.security.component.80.doc/security-component/jsse2Docs/matchsslcontext_tls.html#matchsslcontext_tls。 安装Maven 开发环境的基本配置。用于项目管理，贯穿软件开发生命周期。 7-zip 用于解压“*.zip”和“*.rar”文件,支持7-Zip 16.04版本。

参考信息 针对MapReduce提供的几个样例程序，其对应的依赖包如下： MapReduce统计样例程序 没有需要额外导入的jar包 MapReduce访问多组件样例程序 导入样例工程之后，如果需要使用访问多组件样例程序，请确保集群已安装Hive、HBase服务。 不使用访问多组件样例程序时，如果不影响统计样例程序的正常编译，可忽略多组件样例程序相关报错信息，否则请在导入样例工程后将多组件样例程序类文件删除。

开发流程 工作流配置文件“workflow.xml”（“coordinator.xml”是对工作流进行调度，“bundle.xml”是对一组coordinator进行管理）与“job.properties”。 如果有实现代码，需要开发对应的jar包，例如Java Action；如果是Hive，则需要开发SQL文件。 上传配置文件与jar包（包括依赖的jar包）到HDFS，上传的路径取决于workflow.xml中的oozie.wf.application.path配置的路径。 提供三种方式对工作流进行操作，详情请参见Oozie应用开发常见问题。 Shell命令 Java API Hue Oozie客户端提供了比较完整的examples示例供用户参考，包括各种类型的Action，以及Coordinator以及Bundle的使用。以客户端安装目录为“/opt/client”为例，examples具体目录为“/opt/client/Oozie/oozie-client-*/examples”。 如下通过一个Mapreduce工作流的示例演示如何配置文件，并通过Shell命令调用。

解决步骤 检查工程conf目录下“producer.properties”中配置的“bootstrap.servers”配置值中访问的IP和端口是否正确： 如果IP与Kafka集群部署的业务IP不一致，那么需要修改为当前集群正确的IP地址。 如果配置中的端口为21007（Kafka安全模式端口），那么修改该端口为9092（Kafka普通模式端口）。 检查网络是否正常，确保当前机器能够正常访问Kafka集群。

操作步骤 控制台显示运行结果会有如下成功信息： cluset status is falseWarning: Could not get charToByteConverterClass!Workflow job submitted: 0000067-160729120057089-oozie-omm-WWorkflow job running ...0000067-160729120057089-oozie-omm-WWorkflow job running ...0000067-160729120057089-oozie-omm-WWorkflow job running ...0000067-160729120057089-oozie-omm-WWorkflow job running ...0000067-160729120057089-oozie-omm-WWorkflow job running ...0000067-160729120057089-oozie-omm-WWorkflow job running ...0000067-160729120057089-oozie-omm-WWorkflow job running ...0000067-160729120057089-oozie-omm-WWorkflow job completed ...0000067-160729120057089-oozie-omm-WWorkflow id[0000067-160729120057089-oozie-omm-W] status[SUCCEEDED]-----------finish Oozie ------------------- 同时在HDFS上生成目录“/user/developuser/examples/output-data/map-reduce”，包括如下两个文件： _SUC CES S part-00000 可以通过Hue的文件浏览器或者通过HDFS如下命令行查看： hdfs dfs -ls /user/developuser/examples/output-data/map-reduce 在Windows下面执行的时候可能会出现下面的异常，但是不影响业务： java.io.IOException: Could not locate executable null\bin\winutils.exe in the Hadoop binaries.

Oozie Java接口介绍 Java API主要由org.apache.oozie.client.OozieClient提供。 表1 接口介绍 方法 说明 public String run(Properties conf) 运行Job。 public void start(String jobId) 启动指定的Job。 public String submit(Properties conf) 提交Job。 public void kill(String jobId) 删除指定的Job。 public void suspend(String jobId) 暂停指定的Job。 public void resume(String jobId) 恢复指定的Job。 public WorkflowJob getJobInfo(String jobId) 获取Job信息。 父主题： 常用Oozie API接口介绍

打包项目 通过IDEA自带的Maven工具，打包项目，生成jar包。具体操作请参考在Linux环境中调测Spark应用。 将打包生成的jar包上传到Spark客户端所在服务器的任意目录（例如“$SPARK_HOME” ）下。 若运行“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.JavaHBaseBulkPutExample SparkOnHbaseJavaExample-1.0.jar bulktable cf1 python版本（文件名等请与实际保持一致，此处仅为示例） bin/spark-submit --master yarn --deploy-mode client --jars SparkOnHbaseJavaExample-1.0.jar HBaseBulkPutExample.py bulktable cf1 yarn-cluster模式： java/scala版本（类名等请与实际代码保持一致，此处仅为示例） bin/spark-submit --master yarn --deploy-mode cluster --class com.huawei.bigdata.spark.examples.hbasecontext.JavaHBaseBulkPutExample SparkOnHbaseJavaExample-1.0.jar bulktable cf1 python版本（文件名等请与实际保持一致，此处仅为示例） bin/spark-submit --master yarn --deploy-mode cluster --jars SparkOnHbaseJavaExample-1.0.jar HBaseBulkPutExample.py bulktable cf1

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

Python样例代码 下面代码片段仅为演示，具体代码参见SparkOnHbasePythonExample中HBaseForEachPartitionExample文件： # -*- coding:utf-8 -*-"""【说明】由于pyspark不提供Hbase相关api,本样例使用Python调用Java的方式实现"""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” ）下。 若运行“Spark on HBase”样例程序，需要在Spark客户端的“spark-defaults.conf”配置文件中将配置项“spark.yarn.security.credentials.hbase.enabled”设置为“true”（该参数值默认为“false”，改为“true”后对已有业务没有影响。如果要卸载HBase服务，卸载前请将此参数值改回“false”），将配置项“spark.inputFormat.cache.enabled”设置为“false”。

创建ClickHouse数据库 本章节介绍创建ClickHouse数据库样例代码。 通过on cluster语句在集群中创建表1中以databaseName参数值为数据库名的数据库。 示例代片段参考如下： private void createDatabase(String databaseName, String clusterName) throws Exception { String createDbSql = "create database if not exists " + databaseName + " on cluster " + clusterName; util.exeSql(createDbSql);} 父主题： 开发ClickHouse应用

提交命令 假设用例代码打包后的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-1.0.jar table2 cf1 python版本（文件名等请与实际保持一致，此处仅为示例） bin/spark-submit --master yarn --deploy-mode client --jars SparkOnHbaseJavaExample-1.0.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 SparkOnHbaseJavaExample-1.0.jar table2 cf1 python版本（文件名等请与实际保持一致，此处仅为示例） bin/spark-submit --master yarn --deploy-mode cluster --jars SparkOnHbaseJavaExample-1.0.jar HBaseForEachPartitionExample.py table2 cf1

回答 导致这个问题的主要原因是，yarn-client和yarn-cluster模式在提交任务时setAppName的执行顺序不同导致，yarn-client中setAppName是在向yarn注册Application之前读取，yarn-cluser模式则是在向yarn注册Application之后读取，这就导致yarn-cluster模式设置的应用名不生效。 解决措施： 在spark-submit脚本提交任务时用--name设置应用名和sparkconf.setAppName(appname)里面的应用名一样。 比如代码里设置的应用名为Spark Pi，用yarn-cluster模式提交应用时可以这样设置，在--name后面添加应用名，执行的命令如下： ./spark-submit --class org.apache.spark.examples.SparkPi --master yarn --deploy-mode cluster --name SparkPi jars/original-spark-examples*.jar 10

数据规划 首先需要把原日志文件放置在HDFS系统里。 在Linux系统上新建两个文本文件，将log1.txt中的内容复制保存到input_data1.txt，将log2.txt中的内容复制保存到input_data2.txt。 在HDFS上建立一个文件夹“/tmp/input”，并上传input_data1.txt，input_data2.txt到此目录，命令如下： 登录HDFS客户端。 cd 客户端安装目录 source bigdata_env 在Linux系统HDFS客户端使用命令hdfs dfs -mkdir /tmp/input 在Linux系统HDFS客户端使用命令hdfs dfs -putlocal_filepath /tmp/input

回答 由于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"

查看调试结果 SLF4J: Class path contains multiple SLF4J bindings.SLF4J: Found binding in [jar:file:/D:/mavenlocal/org/apache/logging/log4j/log4j-slf4j-impl/2.6.2/log4j-slf4j-impl-2.6.2.jar!/org/slf4j/impl/StaticLoggerBinder.class]SLF4J: Found binding in [jar:file:/D:/mavenlocal/org/slf4j/slf4j-log4j12/1.7.30/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.SLF4J: Actual binding is of type [org.apache.logging.slf4j.Log4jLoggerFactory]ERROR StatusLogger No log4j2 configuration file found. Using default configuration: logging only errors to the console.---- Begin executing sql: CREATE TABLE IF NOT EXISTS CHILD (NAME STRING, AGE INT) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' ----Result---- Done executing sql: CREATE TABLE IF NOT EXISTS CHILD (NAME STRING, AGE INT) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' -------- Begin executing sql: LOAD DATA INPATH 'hdfs:/data/data' INTO TABLE CHILD ----Result---- Done executing sql: LOAD DATA INPATH 'hdfs:/data/data' INTO TABLE CHILD -------- Begin executing sql: SELECT * FROM child ----NAMEAGEMiranda32Karlie23Candice27---- Done executing sql: SELECT * FROM child -------- Begin executing sql: DROP TABLE child ----Result---- Done executing sql: DROP TABLE child ----Process finished with exit code 0

编包并运行程序 获取样例代码。 下载样例工程的Maven工程源码和配置文件，请参见获取MRS应用开发样例工程。 将样例代码导入IDEA中。 获取配置文件。 从集群的客户端中获取文件。在“$SPARK_HOME/conf”中下载hive-site.xml与spark-defaults.conf文件到本地。 在HDFS中上传数据。 在Liunx中新建文本文件data，将如下数据内容保存到data文件中。 Miranda,32 Karlie,23 Candice,27 在Linux系统HDFS客户端使用命令hadoop fs -mkdir /data（hdfs dfs命令有同样的作用），创建对应目录。 在Linux系统HDFS客户端使用命令hadoop fs -put data /data，上传数据文件。 在样例代码中配置相关参数。 将加载数据的sql语句改为“LOAD DATA INPATH 'hdfs:/data/data' INTO TABLE CHILD”。 在程序运行时添加运行参数，分别为hive-site.xml与spark-defaults.conf文件的路径。 运行程序。

通过Java API提交Oozie作业开发思路 通过典型场景，用户可以快速学习和掌握Oozie的开发过程，并且对关键的接口函数有所了解。 本示例演示了如何通过Java API提交MapReduce作业和查询作业状态，代码示例只涉及了MapReduce作业，其他作业的API调用代码是一样的，只是job配置“job.properties”与工作流配置“workflow.xml”不一样。 完成导入并配置Oozie样例工程操作后即可执行通过Java API提交MapReduce作业和查询作业状态。 父主题： 通过Java API提交Oozie作业

场景说明 假定用户有某个周末网民网购停留时间的日志文本，基于某些业务要求，要求开发MapReduce应用程序实现如下功能： 统计日志文件中本周末网购停留总时间超过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

参数解释 FS Action节点中包含的各参数及其含义，请参见表1。 表1 参数含义 参数 含义 name FS活动的名称 delete 删除指定的文件和目录的标签 move 将文件从源目录移动到目标目录的标签 chmod 修改文件或目录权限的标签 path 当前文件路径 source 源文件路径 target 目标文件路径 permissions 权限字符串 “${变量名}”表示：该值来自“job.properties”所定义。 例如：${nameNode}表示的就是“hdfs://hacluster”。（可参见job.properties）

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

回答 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-clients-2.4.0.jar,$SPARK_CLIENT_HOME/jars/streamingClient010/kafka_2.12-*.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权限用户）。 确保集群安装完成，包括HDFS、Yarn、Spark和Kafka。 本地新建文件“input_data1.txt”，将“log1.txt”的内容复制保存到“input_data1.txt”。 在客户端安装节点下创建文件目录：“/home/data”。将上述文件上传到此“/home/data”目录下。 创建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/*

问题 执行Spark Core应用，尝试收集大量数据到Driver端，当Driver端内存不足时，应用挂起不退出，日志内容如下。 16/04/19 15:56:22 ERROR Utils: Uncaught exception in thread task-result-getter-2java.lang.OutOfMemoryError: Java heap spaceat java.lang.reflect.Array.newArray(Native Method)at java.lang.reflect.Array.newInstance(Array.java:75)at java.io.ObjectInputStream.readArray(ObjectInputStream.java:1671)at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1345)at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2000)at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1924)at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801)at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351)at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2000)at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1924)at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801)at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351)at java.io.ObjectInputStream.readArray(ObjectInputStream.java:1707)at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1345)at java.io.ObjectInputStream.readObject(ObjectInputStream.java:371)at org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:71)at org.apache.spark.serializer.JavaSerializerInstance.deserialize(JavaSerializer.scala:91)at org.apache.spark.scheduler.DirectTaskResult.value(TaskResult.scala:94)at org.apache.spark.scheduler.TaskResultGetter$$anon$3$$anonfun$run$1.apply$mcV$sp(TaskResultGetter.scala:66)at org.apache.spark.scheduler.TaskResultGetter$$anon$3$$anonfun$run$1.apply(TaskResultGetter.scala:57)at org.apache.spark.scheduler.TaskResultGetter$$anon$3$$anonfun$run$1.apply(TaskResultGetter.scala:57)at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:1716)at org.apache.spark.scheduler.TaskResultGetter$$anon$3.run(TaskResultGetter.scala:56)at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)at java.lang.Thread.run(Thread.java:745)Exception in thread "task-result-getter-2" java.lang.OutOfMemoryError: Java heap spaceat java.lang.reflect.Array.newArray(Native Method)at java.lang.reflect.Array.newInstance(Array.java:75)at java.io.ObjectInputStream.readArray(ObjectInputStream.java:1671)at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1345)at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2000)at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1924)at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801)at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351)at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2000)at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1924)at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801)at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351)at java.io.ObjectInputStream.readArray(ObjectInputStream.java:1707)at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1345)at java.io.ObjectInputStream.readObject(ObjectInputStream.java:371)at org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:71)at org.apache.spark.serializer.JavaSerializerInstance.deserialize(JavaSerializer.scala:91)at org.apache.spark.scheduler.DirectTaskResult.value(TaskResult.scala:94)at org.apache.spark.scheduler.TaskResultGetter$$anon$3$$anonfun$run$1.apply$mcV$sp(TaskResultGetter.scala:66)at org.apache.spark.scheduler.TaskResultGetter$$anon$3$$anonfun$run$1.apply(TaskResultGetter.scala:57)at org.apache.spark.scheduler.TaskResultGetter$$anon$3$$anonfun$run$1.apply(TaskResultGetter.scala:57)at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:1716)at org.apache.spark.scheduler.TaskResultGetter$$anon$3.run(TaskResultGetter.scala:56)at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)at java.lang.Thread.run(Thread.java:745)

回答 用户尝试收集大量数据到Driver端，如果Driver端的内存不足以存放这些数据，那么就会抛出OOM(OutOfMemory)的异常，然后Driver端一直在进行GC，尝试回收垃圾来存放返回的数据，导致应用长时间挂起。 解决措施： 如果用户需要在OOM场景下强制将应用退出，那么可以在启动Spark Core应用时，在客户端配置文件“$SPARK_HOME/conf/spark-defaults.conf”中的配置项“spark.driver.extraJavaOptions”中添加如下内容： -XX:OnOutOfMemoryError='kill -9 %p'

基本概念 ResourceManager（RM） RM是一个全局的资源管理器，负责整个系统的资源管理和分配。它主要由两个组件构成：调度器（Scheduler）和应用程序管理器（Applications Manager，ASM）。 ApplicationMaster（AM） 用户提交的每个应用程序均包含一个AM，主要功能包括： 与RM调度器协商以获取资源（用Container表示）。 将得到的资源进一步分配给内部任务。 与NM通信以启动/停止任务。 监控所有任务的运行状态，并在任务运行失败时重新为任务申请资源以重启任务。 NodeManager（NM） NM是每个节点上的资源和任务管理器，一方面，它会定时地向RM汇报本节点上的资源使用情况和各个Container的运行状态；另一方面，它会接收并处理来自AM的Container启动/停止等各种请求。 Container Container是YARN中的资源抽象，它封装了某个节点上的多维度资源，如内存、CPU、磁盘、网络等，当AM向RM申请资源时，RM为AM返回的资源便是用Container表示的。

简介 Yarn是一个分布式的资源管理系统，用于提高分布式的集群环境下的资源利用率，这些资源包括内存、IO、网络、磁盘等。其产生的原因是为了解决原MapReduce框架的不足。最初MapReduce的committer还可以周期性的在已有的代码上进行修改，可是随着代码的增加以及原MapReduce框架设计的不足，在原MapReduce框架上进行修改变得越来越困难，所以MapReduce的committer决定从架构上重新设计MapReduce，使下一代的MapReduce(MRv2/Yarn)框架具有更好的扩展性、可用性、可靠性、向后兼容性和更高的资源利用率，以及能支持除了MapReduce计算框架外的更多的计算框架。