Structured Streaming可靠性说明 Structured Streaming通过checkpoint和WAL机制，对可重放的sources，以及支持重复处理的幂等性sinks，可以提供端到端的exactly-once容错语义。 用户可在程序中设置option("checkpointLocation", "checkpoint路径")启用checkpoint。 从checkpoint恢复时，应用程序或者配置可能发生变更，有部分变更会导致从checkpoint恢复失败，具体限制如下： 不允许source的个数或者类型发生变化。 source的参数变化，这种情况是否能被支持，取决于source类型和查询语句，例如： 速率控制相关参数的添加、删除和修改，此种情况能被支持，如：spark.readStream.format("kafka").option("subscribe", "topic")变更为spark.readStream.format("kafka").option("subscribe", "topic").option("maxOffsetsPerTrigger", ...) 修改消费的topic/files可能会出现不可预知的问题，如：spark.readStream.format("kafka").option("subscribe", "topic")变更为spark.readStream.format("kafka").option("subscribe", "newTopic") sink的类型发生变化：允许特定的几个sink的组合，具体场景需要验证确认，例如： File sink允许变更为kafka sink，kafka中只处理新数据。 kafka sink不允许变更为file sink。 kafka sink允许变更为foreach sink，反之亦然。 sink的参数变化，这种情况是否能被支持，取决于sink类型和查询语句，例如： 不允许file sink的输出路径发生变更。 允许Kafka sink的输出topic发生变更。 允许foreach sink中的自定义算子代码发生变更，但是变更结果取决于用户代码。 Projection、filter和map-like操作变更，局部场景下能够支持，例如： 支持Filter的添加和删除，如：sdf.selectExpr("a")变更为sdf.where(...).selectExpr("a").filter(...) Output schema相同时，projections允许变更，如：sdf.selectExpr("stringColumn AS json").writeStream变更为sdf.select(to_json(...).as("json")).writeStream Output schema不相同时，projections在部分条件下允许变更，如：sdf.selectExpr("a").writeStream变更为sdf.selectExpr("b").writeStream，只有当sink支持“a”到“b”的schema转换时才不会出错。 状态操作的变更，在部分场景下会导致状态恢复失败： Streaming aggregation：如sdf.groupBy("a").agg(...)操作中，不允许分组键或聚合键的类型或者数量发生变化。 Streaming deduplication：如：sdf.dropDuplicates("a")操作中，不允许分组键或聚合键的类型或者数量发生变化。 Stream-stream join：如sdf1.join(sdf2, ...)操作中，关联键的schema不允许发生变化，join类型不允许发生变化，其他join条件的变更可能导致不确定性结果。 任意状态计算：如sdf.groupByKey(...).mapGroupsWithState(...)或者sdf.groupByKey(...).flatMapGroupsWithState(...)操作中，用户自定义状态的schema或者超时类型都不允许发生变化；允许用户自定义state-mapping函数变化，但是变更结果取决于用户代码；如果需要支持schema变更，用户可以将状态数据编码/解码成二进制数据以支持schema迁移。 Source的容错性支持列表 Sources 支持的Options 容错支持 说明 File source path：必填，文件路径 maxFilesPerTrigger：每次trigger最大文件数（默认无限大） latestFirst：是否有限处理新文件（默认值: false） fileNameOnly：是否以文件名作为新文件校验，而不是使用完整路径进行判断（默认值: false） 支持 支持通配符路径，但不支持以逗号分隔的多个路径。 文件必须以原子方式放置在给定的目录中，这在大多数文件系统中可以通过文件移动操作实现。 Socket Source host：连接的节点ip，必填 port：连接的端口，必填 不支持 - Rate Source rowsPerSecond：每秒产生的行数，默认值1 rampUpTime：在达到rowsPerSecond速度之前的上升时间 numPartitions：生成数据行的并行度 支持 - Kafka Source 参见https://spark.apache.org/docs/3.1.1/structured-streaming-kafka-integration.html 支持 - Sink的容错性支持列表 Sinks 支持的output模式 支持Options 容错性 说明 File Sink Append Path：必须指定 指定的文件格式，参见DataFrameWriter中的相关接口 exactly-once 支持写入分区表，按时间分区用处较大 Kafka Sink Append, Update, Complete 参见：https://spark.apache.org/docs/3.1.1/structured-streaming-kafka-integration.html at-least-once 参见https://spark.apache.org/docs/3.1.1/structured-streaming-kafka-integration.html Foreach Sink Append, Update, Complete None 依赖于ForeachWriter实现 参见https://spark.apache.org/docs/3.1.1/structured-streaming-programming-guide.html#using-foreach ForeachBatch Sink Append, Update, Complete None 依赖于算子实现 参见https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#using-foreach-and-foreachbatch Console Sink Append, Update, Complete numRows：每轮打印的行数，默认20 truncate：输出太长时是否清空，默认true 不支持容错 - Memory Sink Append, Complete None 不支持容错，在complete模式下，重启query会重建整个表 -

Structured Streaming支持的功能 支持对流式数据的ETL操作。 支持流式DataFrames或Datasets的schema推断和分区。 流式DataFrames或Datasets上的操作：包括无类型，类似SQL的操作（比如select、where、groupBy），以及有类型的RDD操作（比如map、filter、flatMap）。 支持基于Event Time的聚合计算，支持对迟到数据的处理。 支持对流式数据的去除重复数据操作。 支持状态计算。 支持对流处理任务的监控。 支持批流join，流流join。 当前JOIN操作支持列表如下： 左表 右表 支持的Join类型 说明 Static Static 全部类型 即使在流处理中，不涉及流数据的join操作也能全部支持 Stream Static Inner 支持，但是无状态 Left Outer 支持，但是无状态 Right Outer 不支持 Full Outer 不支持 Stream Stream Inner 支持，左右表可选择使用watermark或者时间范围进行状态清理 Left Outer 有条件的支持，左表可选择使用watermark进行状态清理，右表必须使用watermark+时间范围 Right Outer 有条件的支持，右表可选择使用watermark进行状态清理，左表必须使用watermark+时间范围 Full Outer 不支持

Structured Streaming不支持的功能 不支持多个流聚合。 不支持limit、first、take这些取N条Row的操作。 不支持Distinct。 只有当output mode为complete时才支持排序操作。 有条件地支持流和静态数据集之间的外连接。 不支持部分DataSet上立即运行查询并返回结果的操作： count()：无法从流式Dataset返回单个计数，而是使用ds.groupBy().count()返回一个包含运行计数的streaming Dataset。 foreach()：使用ds.writeStream.foreach(...)代替。 show()：使用输出console sink代替。

Structured Streaming可靠性说明 Structured Streaming通过checkpoint和WAL机制，对可重放的sources，以及支持重复处理的幂等性sinks，可以提供端到端的exactly-once容错语义。 用户可在程序中设置option("checkpointLocation", "checkpoint路径")启用checkpoint。 从checkpoint恢复时，应用程序或者配置可能发生变更，有部分变更会导致从checkpoint恢复失败，具体限制如下： 不允许source的个数或者类型发生变化。 source的参数变化，这种情况是否能被支持，取决于source类型和查询语句，例如： 速率控制相关参数的添加、删除和修改，此种情况能被支持，如：spark.readStream.format("kafka").option("subscribe", "topic")变更为spark.readStream.format("kafka").option("subscribe", "topic").option("maxOffsetsPerTrigger", ...) 修改消费的topic/files可能会出现不可预知的问题，如：spark.readStream.format("kafka").option("subscribe", "topic")变更为spark.readStream.format("kafka").option("subscribe", "newTopic") sink的类型发生变化：允许特定的几个sink的组合，具体场景需要验证确认，例如： File sink允许变更为kafka sink，kafka中只处理新数据。 kafka sink不允许变更为file sink。 kafka sink允许变更为foreach sink，反之亦然。 sink的参数变化，这种情况是否能被支持，取决于sink类型和查询语句，例如： 不允许file sink的输出路径发生变更。 允许Kafka sink的输出topic发生变更。 允许foreach sink中的自定义算子代码发生变更，但是变更结果取决于用户代码。 Projection、filter和map-like操作变更，局部场景下能够支持，例如： 支持Filter的添加和删除，如：sdf.selectExpr("a")变更为sdf.where(...).selectExpr("a").filter(...) Output schema相同时，projections允许变更，如：sdf.selectExpr("stringColumn AS json").writeStream变更为sdf.select(to_json(...).as("json")).writeStream Output schema不相同时，projections在部分条件下允许变更，如：sdf.selectExpr("a").writeStream变更为sdf.selectExpr("b").writeStream，只有当sink支持“a”到“b”的schema转换时才不会出错。 状态操作的变更，在部分场景下会导致状态恢复失败： Streaming aggregation：如sdf.groupBy("a").agg(...)操作中，不允许分组键或聚合键的类型或者数量发生变化。 Streaming deduplication：如：sdf.dropDuplicates("a")操作中，不允许分组键或聚合键的类型或者数量发生变化。 Stream-stream join：如sdf1.join(sdf2, ...)操作中，关联键的schema不允许发生变化，join类型不允许发生变化，其他join条件的变更可能导致不确定性结果。 任意状态计算：如sdf.groupByKey(...).mapGroupsWithState(...)或者sdf.groupByKey(...).flatMapGroupsWithState(...)操作中，用户自定义状态的schema或者超时类型都不允许发生变化；允许用户自定义state-mapping函数变化，但是变更结果取决于用户代码；如果需要支持schema变更，用户可以将状态数据编码/解码成二进制数据以支持schema迁移。 Source的容错性支持列表 Sources 支持的Options 容错支持 说明 File source path：必填，文件路径 maxFilesPerTrigger：每次trigger最大文件数（默认无限大） latestFirst：是否有限处理新文件（默认值: false） fileNameOnly：是否以文件名作为新文件校验，而不是使用完整路径进行判断（默认值: false） 支持 支持通配符路径，但不支持以逗号分隔的多个路径。 文件必须以原子方式放置在给定的目录中，这在大多数文件系统中可以通过文件移动操作实现。 Socket Source host：连接的节点ip，必填 port：连接的端口，必填 不支持 - Rate Source rowsPerSecond：每秒产生的行数，默认值1 rampUpTime：在达到rowsPerSecond速度之前的上升时间 numPartitions：生成数据行的并行度 支持 - Kafka Source 参见https://spark.apache.org/docs/3.1.1/structured-streaming-kafka-integration.html 支持 - Sink的容错性支持列表 Sinks 支持的output模式 支持Options 容错性 说明 File Sink Append Path：必须指定 指定的文件格式，参见DataFrameWriter中的相关接口 exactly-once 支持写入分区表，按时间分区用处较大 Kafka Sink Append, Update, Complete 参见：https://spark.apache.org/docs/3.1.1/structured-streaming-kafka-integration.html at-least-once 参见https://spark.apache.org/docs/3.1.1/structured-streaming-kafka-integration.html Foreach Sink Append, Update, Complete None 依赖于ForeachWriter实现 参见https://spark.apache.org/docs/3.1.1/structured-streaming-programming-guide.html#using-foreach ForeachBatch Sink Append, Update, Complete None 依赖于算子实现 参见https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#using-foreach-and-foreachbatch Console Sink Append, Update, Complete numRows：每轮打印的行数，默认20 truncate：输出太长时是否清空，默认true 不支持容错 - Memory Sink Append, Complete None 不支持容错，在complete模式下，重启query会重建整个表 -

Structured Streaming支持的功能 支持对流式数据的ETL操作。 支持流式DataFrames或Datasets的schema推断和分区。 流式DataFrames或Datasets上的操作：包括无类型，类似SQL的操作（比如select、where、groupBy），以及有类型的RDD操作（比如map、filter、flatMap）。 支持基于Event Time的聚合计算，支持对迟到数据的处理。 支持对流式数据的去除重复数据操作。 支持状态计算。 支持对流处理任务的监控。 支持批流join，流流join。 当前JOIN操作支持列表如下： 左表 右表 支持的Join类型 说明 Static Static 全部类型 即使在流处理中，不涉及流数据的join操作也能全部支持 Stream Static Inner 支持，但是无状态 Left Outer 支持，但是无状态 Right Outer 不支持 Full Outer 不支持 Stream Stream Inner 支持，左右表可选择使用watermark或者时间范围进行状态清理 Left Outer 有条件的支持，左表可选择使用watermark进行状态清理，右表必须使用watermark+时间范围 Right Outer 有条件的支持，右表可选择使用watermark进行状态清理，左表必须使用watermark+时间范围 Full Outer 不支持

Structured Streaming不支持的功能 不支持多个流聚合。 不支持limit、first、take这些取N条Row的操作。 不支持Distinct。 只有当output mode为complete时才支持排序操作。 有条件地支持流和静态数据集之间的外连接。 不支持部分DataSet上立即运行查询并返回结果的操作： count()：无法从流式Dataset返回单个计数，而是使用ds.groupBy().count()返回一个包含运行计数的streaming Dataset。 foreach()：使用ds.writeStream.foreach(...)代替。 show()：使用输出console sink代替。

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

解决方案 提交yarn-client模式的结构流任务时需要额外如下操作： 将Spark客户端目录下spark-default.conf文件中的spark.driver.extraClassPath配置复制出来，并将Kafka相关jar包路径追加到该配置项之后，提交结构流任务时需要通过--conf将该配置项给加上。例如：kafka相关jar包路径为“/kafkadir”，提交任务需要增加--conf spark.driver.extraClassPath=/opt/client/Spark2x/spark/conf/:/opt/client/Spark2x/spark/jars/*:/opt/client/Spark2x/spark/x86/*:/kafkadir/*。 提交yarn-cluster模式的结构流任务时需要额外如下操作： 将Spark客户端目录下spark-default.conf文件中的spark.yarn.cluster.driver.extraClassPath配置给复制出来，并将Kafka相关jar包相对路径追加到该配置项之后，提交结构流任务时需要通过--conf 将该配置项给加上。例如：kafka相关包为kafka-clients-x.x.x.jar，kafka_2.11-x.x.x.jar，提交任务需要增加--conf spark.yarn.cluster.driver.extraClassPath=/home/huawei/Bigdata/common/runtime/security:./kafka-clients-x.x.x.jar:./kafka_2.11-x.x.x.jar。 当前版本Spark结构流部分不再支持kafka2.x之前的版本，对于升级场景请继续使用旧的客户端。

回答 第三方jar包（例如自定义udf）区分x86和TaiShan版本时，混合使用方案： 进入到服务端Spark2x SparkResource的安装目录（集群安装时，SparkResource可能会安装在多个节点上，登录任意一个SparkResource节点，进入到SparkResource的安装目录）。 准备好自己的jar包，例如xx.jar的x86版本和TaiShan版本。将x86版本和TaiShan版本的xx.jar分别复制到当前目录的x86文件夹和TaiShan文件夹里面。 在当前目录下执行以下命令将jar包打包： zip -qDj spark-archive-2x-x86.zip x86/* zip -qDj spark-archive-2x-arm.zip arm/* 执行以下命令查看hdfs上的spark2x依赖的jar包： hdfs dfs -ls /user/spark2x/jars/8.1.0.1 8.1.0.1是版本号，不同版本不同。 执行以下命令移动hdfs上旧的jar包文件到其他目录，例如移动到“tmp”目录。 hdfs dfs -mv /user/spark2x/jars/8.1.0.1/spark-archive-2x-arm.zip /tmp hdfs dfs -mv /user/spark2x/jars/8.1.0.1/spark-archive-2x-x86.zip /tmp 上传3中打包的spark-archive-2x-arm.zip和spark-archive-2x-x86.zip到hdfs的/user/spark2x/jars/8.1.0.1目录下，上传命令如下： hdfs dfs -put spark-archive-2x-arm.zip /user/spark2x/jars/8.1.0.1/ hdfs dfs -put spark-archive-2x-x86.zip /user/spark2x/jars/8.1.0.1/ 上传完毕后删除本地的spark-archive-2x-arm.zip，spark-archive-2x-x86.zip文件。 对其他的sparkResource安装节点执行1~2。 进入webUI重启spark2x的jdbcServer实例。 重启后，需要更新客户端配置。按照客户端所在的机器类型（x86、TaiShan）复制xx.jar的相应版本到客户端的Spark2x安装目录“${install_home}/Spark2x/spark/jars”文件夹中。${install_home}是用户的客户端安装路径，用户需要填写实际的安装目录；若本地的安装目录为“/opt/hadoopclient”，那么就复制相应版本xx.jar到“/opt/hadoopclient/Spark2x/spark/jars”文件夹里。

问题 使用运行的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

回答 如下图所示，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参数值根据应用场景，适当调整。

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

回答 第三方jar包（例如自定义udf）区分x86和TaiShan版本时，混合使用方案： 进入到服务端spark2x sparkResource的安装目录（这个集群安装过程中可能会安装在多个节点上，随便进入一个安装节点，cd到sparkResource的安装目录）。 准备好自己的jar包例如xx.jar的x86版本和TaiShan版本。将x86版本和TaiShan版本的xx.jar分别复制到当前目录的x86文件夹和TaiShan文件夹里面。 在当前目录下执行以下命令将jar包打包： zip -qDj spark-archive-2x-x86.zip x86/* zip -qDj spark-archive-2x-arm.zip arm/* 执行以下命令查看hdfs上的spark2x依赖的jar包： hdfs dfs -ls /user/spark2x/jars/8.1.0.1 8.1.0.1是版本号，不同版本不同。 执行以下命令移动hdfs上旧的jar包文件到其他目录，例如移动到“tmp”目录。 hdfs dfs -mv /user/spark2x/jars/8.1.0.1/spark-archive-2x-arm.zip /tmp hdfs dfs -mv /user/spark2x/jars/8.1.0.1/spark-archive-2x-x86.zip /tmp 上传3中打包的spark-archive-2x-arm.zip和spark-archive-2x-x86.zip到hdfs的/user/spark2x/jars/8.1.0.1目录下，上传命令如下： hdfs dfs -put spark-archive-2x-arm.zip /user/spark2x/jars/8.1.0.1/ hdfs dfs -put spark-archive-2x-x86.zip /user/spark2x/jars/8.1.0.1/ 上传完毕后删除本地的spark-archive-2x-arm.zip，spark-archive-2x-x86.zip文件。 对其他的SparkResource安装节点执行1~2。 进入WebUI重启Spark的JD BCS erver实例。 重启后，需要更新客户端配置。按照客户端所在的机器类型（x86、TaiShan）复制xx.jar的相应版本到客户端的spark2x安装目录${install_home}/Spark2x/spark/jars文件夹中。${install_home}是用户的客户端安装路径，用户需要填写实际的安装目录；若本地的安装目录为/opt/hadoopclient,那么就复制相应版本xx.jar到/opt/hadoopclient/Spark2x/spark/jars文件夹里。

问题 执行Spark Core应用，尝试收集大量数据到Driver端，当Driver端内存不足时，应用挂起不退出，日志内容如下。 16/04/19 15:56:22 ERROR Utils: Uncaught exception in thread task-result-getter-2 java.lang.OutOfMemoryError: Java heap space at 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 space at 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'