华为云用户手册

回答 这是正常现象。 数据分到哪个partition是通过对key的hashcode取模得到的，不同的hashcode取模后的结果有可能是一样的，那样数据就会被分到相同的partition里面，因此出现有些partition没有数据而有些partition里面有多个key对应的数据。 通过调整“spark.sql.shuffle.partitions”参数值可以调整取模时的基数，改善数据分块不均匀的情况，多次验证发现配置为质数或者奇数效果比较好。 在Driver端的“spark-defaults.conf”配置文件中调整如下参数。 表1 参数说明 参数 描述 默认值 spark.sql.shuffle.partitions shuffle操作时，shuffle数据的分块数。 200

问题 在Spark2x的spark-shell上执行如下代码失败： val acctId = List(("49562", "Amal", "Derry"), ("00000", "Fred", "Xanadu"))val rddLeft = sc.makeRDD(acctId)val dfLeft = rddLeft.toDF("Id", "Name", "City")//dfLeft.showval acctCustId = List(("Amal", "49562", "CO"), ("Dave", "99999", "ZZ"))val rddRight = sc.makeRDD(acctCustId)val dfRight = rddRight.toDF("Name", "CustId", "State")//dfRight.showval dfJoin = dfLeft.join(dfRight, dfLeft("Id") === dfRight("CustId"), "outer")dfJoin.showdfJoin.repartition(1).write.format("com.databricks.spark.csv").option("delimiter", "\t").option("header", "true").option("treatEmptyValuesAsNulls", "true").option("nullValue", "").save("/tmp/outputDir")

回答 当userB对tableB执行insert操作后，会在外表数据路径下生成新的数据文件，且文件属组是userB，当userA查询tableA时，会读取外表数据目录下的所有的文件，此时会因没有userB生成的文件的读取权限而查询失败。 实际上，不只是查询场景，还有其他场景也会出现问题。例如：inset overwrite操作将会把此目录下的其他表文件也一起复写。 由于Spark SQL当前的实现机制，如果对此种场景添加检查限制，会存在一致性问题和性能问题，因此未对此种场景添加限制，但是用户应避免此种用法，以避免此场景带来的各种问题。

解答 该应用程序中使用了DStream中的print算子来显示结果，该算子会调用RDD中的take算子来实现底层的计算。 Take算子会以Partition为单位多次触发计算。 在该问题中，由于Shuffle操作，导致take算子默认有两个Partition，Spark首先计算第一个Partition，但由于没有数据输入，导致获取结果不足10个，从而触发第二次计算，因此会出现RDD的DAG结构打印两次的现象。 在代码中将print算子修改为foreach(collect)，该问题则不会出现。

回答 问题原因： ApplicationMaster进程中有1个Credential Refresh Thread会根据token renew周期 * 0.75的时间比例上传更新后的Credential文件到HDFS上。 Executor进程中有1个Credential Refresh Thread会根据token renew周期 *0.8的时间比例去HDFS上获取更新后的Credential文件，用来刷新UserGroupInformation中的token，避免token失效。 当Executor进程的Credential Refresh Thread发现当前时间已经超过Credential文件更新时间（即token renew周期 *0.8）时，会等待1分钟再去HDFS上面获取最新的Credential文件，以确保AM端已经将更新后的Credential文件放到HDFS上。 当“dfs.namenode.delegation.token.renew-interval”配置值小于60秒，Executor进程起来时发现当前时间已经超过Credential文件更新时间，等待1分钟再去HDFS上面获取最新的Credential文件，而此时token已经失效，task运行失败，然后在其他Executor上重试，由于重试时间都是在1分钟内完成，所以task在其他Executor上也运行失败，导致运行失败的Executor加入到黑名单，没有可用的Executor，应用退出。 修改方案： 在Spark使用场景下，需设置“dfs.namenode.delegation.token.renew-interval”大于80秒。“dfs.namenode.delegation.token.renew-interval”参数描述请参表1考。 表1 参数说明 参数 描述 默认值 dfs.namenode.delegation.token.renew-interval 该参数为服务器端参数，设置token renew的时间间隔，单位为毫秒。 86400000

回答 在executor核数等于1的情况下，遵循以下规则对调优Spark Streaming运行参数有所帮助。 Spark任务处理速度和Kafka上partition个数有关，当partition个数小于给定executor个数时，实际使用的executor个数和partition个数相同，其余的将会被空闲。所以应该使得executor个数小于或者等于partition个数。 当Kafka上不同partition数据有倾斜时，数据较多的partition对应的executor将成为数据处理的瓶颈，所以在执行Producer程序时，数据平均发送到每个partition可以提升处理的速度。 在partition数据均匀分布的情况下，同时提高partition和executor个数，将会提升Spark处理速度（当partition个数和executor个数保持一致时，处理速度是最快的）。 在partition数据均匀分布的情况下，尽量保持partition个数是executor个数的整数倍，这样将会使资源得到合理利用。

问题 使用默认配置时，16T的文本数据转成4T Parquet数据失败，报如下错误信息。 Job aborted due to stage failure: Task 2866 in stage 11.0 failed 4 times, most recent failure: Lost task 2866.6 in stage 11.0 (TID 54863, linux-161, 2): java.io.IOException: Failed to connect to /10.16.1.11:23124at org.apache.spark.network.client.TransportClientFactory.createClient(TransportClientFactory.java:214)at org.apache.spark.network.client.TransportClientFactory.createClient(TransportClientFactory.java:167)at org.apache.spark.network.netty.NettyBlockTransferService$$anon$1.createAndStart(NettyBlockTransferService.scala:92) 使用的默认配置如表1所示。 表1 参数说明 参数 描述 默认值 spark.sql.shuffle.partitions shuffle操作时，shuffle数据的分块数。 200 spark.shuffle.sasl.timeout shuffle操作时SASL认证的超时时间。单位：秒。 120s spark.shuffle.io.connectionTimeout shuffle操作时连接远程节点的超时时间。单位：秒。 120s spark.network.timeout 所有涉及网络连接操作的超时时间。单位：秒。 360s

回答 当前在默认配置下，在内存中保留的Job和Stage的UI数据个数为1000个。 当前大集群优化已增加将UI数据溢出到磁盘的优化，其溢出条件是每个Stage中的UI数据大小达到最小阈值5MB。如果每个Stage的task数较小，那么其UI数据大小可能达不到该阈值，从而导致该Stage的UI数据一直缓存在内存中，直到UI数据个数到达保留的上限值（当前默认值为1000个），旧的UI数据才会在内存中被清除。 因此，在将旧的UI数据从内存中清除之前，UI数据会占用大量内存，从而导致执行10T的TPCDS测试套时出现Driver内存不足的现象。 规避措施： 根据业务需要，配置合适的需要保留的Job和Stage的UI数据个数，即配置“spark.ui.retainedJobs”和“spark.ui.retainedStages”参数。详细信息请参考常用参数中的表13。 如果需要保留的Job和Stage的UI数据个数较多，可通过配置“spark.driver.memory”参数，适当增大Driver的内存。详细信息请参考常用参数中的表10。

回答 当前JD BCS erver中存在两个线程池HiveServer2-Handler-Pool和HiveServer2-Background-Pool，其中HiveServer2-Handler-Pool用于处理session连接，HiveServer2-Background-Pool用于处理SQL语句的执行。 当前的健康检查机制是通过新建session连接，并在该session所在的线程中执行健康检查命令HEALTHCHECK来判断Spark JDB CS erver的健康状况，因此HiveServer2-Handler-Pool必须保留一个线程，用于处理健康检查的session连接和健康检查命令执行，否则将导致无法建立健康检查的session连接或健康检查命令无法执行，从而认为Spark JDBCServer不健康而被Kill。即如果当前HiveServer2-Handler-Pool的线程池数为100，那么最多支持连接99个session。

问题 执行大数据量的Spark任务（如2T的TPCDS测试套），任务运行成功后，在spark-sql退出时概率性出现RejectedExecutionException的异常栈信息，相关日志如下所示： 16/07/16 10:19:56 ERROR TransportResponseHandler: Still have 2 requests outstanding when connection from linux-192/10.1.1.5:59250 is closedjava.util.concurrent.RejectedExecutionException: Task scala.concurrent.impl.CallbackRunnable@5fc1ab rejected from java.util.concurrent.ThreadPoolExecutor@52fa7e19[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 3025]

问题 当执行一个很复杂的SQL语句时，例如有多层语句嵌套，且单层语句中对字段有大量的逻辑处理（如多层嵌套的case when语句），此时执行该语句会报如下所示的错误日志，该错误表明某个方法的代码超出了64KB。 java.util.concurrent.ExecutionException: java.lang.Exception: failed to compile: org.codehaus.janino.JaninoRuntimeException: Code of method "(Lorg/apache/spark/sql/catalyst/expressions/GeneratedClass$SpecificUnsafeProjection;Lorg/apache/spark/sql/catalyst/InternalRow;)V" of class "org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection" grows beyond 64 KB

问题 Spark执行应用时上报如下类似错误并导致应用结束。 2016-04-20 10:42:00,557 | ERROR | [shuffle-server-2] | Connection to 10-91-8-208/10.18.0.115:57959 has been quiet for 180000 ms while there are outstanding requests. Assuming connection is dead; please adjust spark.network.timeout if this is wrong. | org.apache.spark.network.server.TransportChannelHandler.userEventTriggered(TransportChannelHandler.java:128)2016-04-20 10:42:00,558 | ERROR | [shuffle-server-2] | Still have 1 requests outstanding when connection from 10-91-8-208/10.18.0.115:57959 is closed | org.apache.spark.network.client.TransportResponseHandler.channelUnregistered(TransportResponseHandler.java:102)2016-04-20 10:42:00,562 | WARN | [yarn-scheduler-ask-am-thread-pool-160] | Error sending message [message = DoShuffleClean(application_1459995017785_0108,319)] in 1 attempts | org.apache.spark.Logging$class.logWarning(Logging.scala:92)java.io.IOException: Connection from 10-91-8-208/10.18.0.115:57959 closed at org.apache.spark.network.client.TransportResponseHandler.channelUnregistered(TransportResponseHandler.java:104) at org.apache.spark.network.server.TransportChannelHandler.channelUnregistered(TransportChannelHandler.java:94) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelUnregistered(AbstractChannelHandlerContext.java:158) at io.netty.channel.AbstractChannelHandlerContext.fireChannelUnregistered(AbstractChannelHandlerContext.java:144) at io.netty.channel.ChannelInboundHandlerAdapter.channelUnregistered(ChannelInboundHandlerAdapter.java:53) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelUnregistered(AbstractChannelHandlerContext.java:158) at io.netty.channel.AbstractChannelHandlerContext.fireChannelUnregistered(AbstractChannelHandlerContext.java:144) at io.netty.channel.ChannelInboundHandlerAdapter.channelUnregistered(ChannelInboundHandlerAdapter.java:53) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelUnregistered(AbstractChannelHandlerContext.java:158) at io.netty.channel.AbstractChannelHandlerContext.fireChannelUnregistered(AbstractChannelHandlerContext.java:144) at io.netty.channel.ChannelInboundHandlerAdapter.channelUnregistered(ChannelInboundHandlerAdapter.java:53) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelUnregistered(AbstractChannelHandlerContext.java:158) at io.netty.channel.AbstractChannelHandlerContext.fireChannelUnregistered(AbstractChannelHandlerContext.java:144) at io.netty.channel.DefaultChannelPipeline.fireChannelUnregistered(DefaultChannelPipeline.java:739) at io.netty.channel.AbstractChannel$AbstractUnsafe$8.run(AbstractChannel.java:659) at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:357) at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:357) at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:111) at java.lang.Thread.run(Thread.java:745)2016-04-20 10:42:00,573 | INFO | [dispatcher-event-loop-14] | Starting task 177.0 in stage 1492.0 (TID 1996351, linux-254, PRO CES S_LOCAL, 2106 bytes) | org.apache.spark.Logging$class.logInfo(Logging.scala:59)2016-04-20 10:42:00,574 | INFO | [task-result-getter-0] | Finished task 85.0 in stage 1492.0 (TID 1996259) in 191336 ms on linux-254 (106/3000) | org.apache.spark.Logging$class.logInfo(Logging.scala:59)2016-04-20 10:42:00,811 | ERROR | [Yarn application state monitor] | Yarn application has already exited with state FINISHED! | org.apache.spark.Logging$class.logError(Logging.scala:75)

回答 原因分析： 这是由于Spark2x与Spark1.5存储DataSoure表信息的格式不一致导致的。Spark1.5会将schema信息分成多个part，使用path.park.0作为key进行存储，读取时再将各个part都读取出来，重新拼成完整的信息。而Spark2x直接使用相应的key获取对应的信息。这样在Spark2x中去读取Spark1.5创建的DataSource表时，就无法成功读取到key对应的信息，导致解析DataSource表信息失败。 而在处理Hive格式的表时，Spark2x与Spark1.5的存储方式一致，所以Spark2x可以直接读取Spark1.5创建的表，不存在上述问题。 规避措施： Spark2x可以通过创建外表的方式来创建一张指向Spark1.5表实际数据的表，这样可以实现在Spark2x中读取Spark1.5创建的DataSource表。同时，Spark1.5更新过数据后，Spark2x中访问也能感知到变化 ，反过来一样。这样即可实现Spark2x对Spark1.5创建的DataSource表的访问。

回答 在开启钨丝计划（即tungsten功能）后，Spark对于部分执行计划会使用codegen的方式来生成Java代码，但JDK编译时要求Java代码中的每个函数的长度不能超过64KB。当执行一个很复杂的SQL语句时，例如有多层语句嵌套，且单层语句中对字段有大量的逻辑处理（如多层嵌套的case when语句），这种情况下，通过codegen生成的Java代码中函数的大小就可能会超过64KB，从而导致编译失败。 规避措施： 当出现上述问题时，用户可以通过关闭钨丝计划，关闭使用codegen的方式来生成Java代码的功能，从而确保语句的正常执行。即在客户端的“spark-defaults.conf”配置文件中将“spark.sql.codegen.wholeStage”配置为“false”。

问题 当创建了表名为table的表后，执行drop table table上报以下错误，或者执行其他操作也会出现类似错误。 16/07/12 18:56:29 ERROR SparkSQLDriver: Failed in [drop table table]java.lang.RuntimeException: [1.1] failure: identifier expectedtable^at scala.sys.package$.error(package.scala:27)at org.apache.spark.sql.catalyst.SqlParserTrait$class.parseTableIdentifier(SqlParser.scala:56)at org.apache.spark.sql.catalyst.SqlParser$.parseTableIdentifier(SqlParser.scala:485)

回答 当前可以通过以下3种方式创建UDF： 在Hive端创建UDF。 通过JDBCServer接口创建UDF。用户可以通过Spark Beeline或者JDBC客户端代码来连接JDBCServer，从而执行SQL命令，创建UDF。 通过spark-sql创建UDF。 删除UDF失败，存在以下两种场景： 在Spark Beeline中，对于其他方式创建的UDF，需要重新启动Spark服务端的JDBCServer后，才能将此类UDF删除成功，否则删除失败。在spark-sql中，对于其他方式创建的UDF，需要重新启动spark-sql后，才能将此类UDF删除成功，否则删除失败。 原因：创建UDF后，Spark服务端的JDBCServer未重启或者spark-sql未重新启动的场景，Spark所在线程的FunctionRegistry对象未保存新创建的UDF，那么删除UDF时就会出现错误。 解决方法：重启Spark服务端的JDBCServer和spark-sql，再删除此类UDF。 在Hive端创建UDF时未在创建语句中指定jar包路径，而是通过add jar命令添加UDF的jar包如add jar /opt/test/two_udfs.jar，这种场景下，在其他服务中删除UDF时就会出现ClassNotfound的错误，从而导致删除失败。 原因：在删除UDF时，会先获取该UDF，此时会去加载该UDF对应的类，由于创建UDF时是通过add jar命令指定jar包路径的，其他服务进程的classpath不存在这些jar包，因此会出现ClassNotfound的错误从而导致删除失败。 解决方法：该方式创建的UDF不支持通过其他方式删除，只能通过与创建时一致的方式删除。

回答 Spark SQL对用户SQL语句的执行逻辑是：首先解析出语句中包含的表，再获取表的元数据信息，然后对权限进行检查。 当表是parquet表时，元数据信息包括文件的Split信息。Split信息需要调用HDFS的接口去读取，当表包含的文件数量很多时，串行读取Split信息变得缓慢，影响性能。故对此做了优化，当表包含的文件大于一定阈值（即spark.sql.sources.parallelSplitDiscovery.threshold参数值）时，会生成一个Job，利用Executor的并行能力去读取，从而提升执行效率。 由于权限检查在获取表元数据之后，因此当读取的parquet表包含的文件数量很多时，会在报“Missing Privileges”之前，运行一个Job来并行读取元数据信息。

回答 在进行rollup和cube操作时，用户通常是基于维度进行分析，需要的是度量的结果，因此不会对维度进行聚合操作。 例如当前有表src(d1, d2, m)，那么语句1“select d1, sum(m) from src group by d1, d2 with rollup”就是对维度d1和d2进行上卷操作计算度量m的结果，因此有实际业务意义，而其结果也跟预期是一致的。但语句2“select d1, sum(d1) from src group by d1, d2 with rollup”则从业务上无法解释。当前对于语句2所有聚合（sum/avg/max/min）结果均为0。 只有在rollup和cube操作中对出现在group by中的字段进行聚合结果才是0，非rollup和cube操作其结果跟预期一致。

回答 由于Spark存在一个机制，为了提高性能会缓存Parquet的元数据信息。当通过Hive或其他方式更新了Parquet表时，缓存的元数据信息未更新，导致Spark SQL查询不到新插入的数据。 对于存储类型为Parquet的Hive分区表，在执行插入数据操作后，如果分区信息未改变，则缓存的元数据信息未更新，导致Spark SQL查询不到新插入的数据。 解决措施：在使用Spark SQL查询之前，需执行Refresh操作更新元数据信息。 REFRESH TABLE table_name; table_name为刷新的表名，该表必须存在，否则会出错。 执行查询语句时，即可获取到最新插入的数据。 Spark官网提供了此机制的描述，详情请参见：https://spark.apache.org/docs/3.1.1/sql-programming-guide.html#metadata-refreshing

回答 Spark SQL可以将表cache到内存中，并且使用压缩存储来尽量减少内存压力。通过将表cache，查询可以直接从内存中读取数据，从而减少读取磁盘带来的内存开销。 但需要注意的是，被cache的表会占用executor的内存。尽管在Spark SQL采用压缩存储的方式来尽量减少内存开销、缓解GC压力，但当缓存的表较大或者缓存表数量较多时，将不可避免的影响executor的稳定性。 此时的最佳实践是，当不需要将表cache来实现查询加速时，应及时将表进行uncache以释放内存。可以执行命令uncache table table_name来uncache表。 被cache的表也可以在Spark Driver UI的Storage标签里查看。

问题 向动态分区表中插入数据时，shuffle过程中大面积shuffle文件损坏（磁盘掉线、节点故障等）后，为什么会在重试的task中出现"Failed to CREATE_FILE"异常？ 2016-06-25 15:11:31,323 | ERROR | [Executor task launch worker-0] | Exception in task 15.0 in stage 10.1 (TID 1258) | org.apache.spark.Logging$class.logError(Logging.scala:96)org.apache.hadoop.hive.ql.metadata.HiveException: org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.hdfs.protocol.AlreadyBeingCreatedException): Failed to CREATE_FILE /user/hive/warehouse/testdb.db/web_sales/.hive-staging_hive_2016-06-25_15-09-16_999_8137121701603617850-1/-ext-10000/_temporary/0/_temporary/attempt_201606251509_0010_m_000015_0/ws_sold_date=1999-12-17/part-00015 for DFSClient_attempt_201606251509_0010_m_000015_0_353134803_151 on 10.1.1.5 because this file lease is currently owned by DFSClient_attempt_201606251509_0010_m_000015_0_-848353830_156 on 10.1.1.6

问题 在执行大数据量的Spark任务（如100T的TPCDS测试套）过程中，有时会出现Executor丢失从而导致Stage重试的现象。查看Executor的日志，出现“Executor 532 is lost rpc with driver,but is still alive, going to kill it”所示信息，表明Executor丢失是由于JVM Crash导致的。 JVM的关键Crash错误日志，如下： ## A fatal error has been detected by the Java Runtime Environment:## Internal Error (sharedRuntime.cpp:834), pid=241075, tid=140476258551552# fatal error: exception happened outside interpreter, nmethods and vtable stubs at pc 0x00007fcda9eb8eb1

回答 动态分区表插入数据的最后一步是读取shuffle文件的数据，再写入到表对应的分区文件中。 当大面积shuffle文件损坏后，会引起大批量task失败，然后进行job重试。重试前Spark会将写表分区文件的句柄关闭，大批量task关闭句柄时HDFS无法及时处理。在task进行下一次重试时，句柄在NameNode端未被及时释放，即会抛出"Failed to CREATE_FILE"异常。 这种现象仅会在大面积shuffle文件损坏时发生，出现异常后task会重试，重试耗时在毫秒级，影响较小，可以忽略不计。

问题 Spark应用执行过程中，当driver连接RM失败时，会报下面的错误，且较长时间不退出。 16/04/23 15:31:44 INFO RetryInvocationHandler: Exception while invoking getApplicationReport of class ApplicationClientProtocolPBClientImpl over 37 after 1 fail over attempts. Trying to fail over after sleeping for 44160ms.java.net.ConnectException: Call From vm1/192.168.39.30 to vm1:8032 failed on connection exception: java.net.ConnectException: Connection refused; For more details see: http://wiki.apache.org/hadoop/ConnectionRefused

回答 这是因为ResourceManager感知到NodeManager关闭时，Executor(s)已经因空闲超时而被driver请求kill掉，但因NodeManager已经关闭，这些Executor(s)实际上并不能被kill掉，因此driver不能感知到这些Executor(s)的LOST事件，所以并未从自身的Executor list中移除，从而导致在driver页面上还能看到这些Executor(s)，这是YARN NodeManager关闭之后的正常现象，NodeManager再次启动后，这些Executor(s)会被移除。

回答 对于Hash shuffle，在shuffle的过程中写数据时不做排序操作，只是将数据根据Hash的结果，将各个reduce分区的数据写到各自的磁盘文件中。 这样带来的问题是如果reduce分区的数量比较大的话，将会产生大量的磁盘文件（比如：该问题中将产生1000000 * 100000 = 10^11个shuffle文件）。如果磁盘文件数量特别巨大，对文件读写的性能会带来比较大的影响，此外由于同时打开的文件句柄数量多，序列化以及压缩等操作需要占用非常大的临时内存空间，对内存的使用和GC带来很大的压力，从而容易造成Executor无法响应Driver。 因此，建议使用Sort shuffle，而不使用Hash shuffle。

回答 JDBCServer方式使用了ShuffleService功能，Reduce阶段所有的Executor会从NodeManager中获取数据，当数据量达到一个级别（10T级别），会出现NodeManager单点瓶颈（ShuffleService服务在NodeManager进程中），就会出现某些Task获取数据超时，从而出现该问题。 因此，当数据量达到10T级别以上的Spark任务，建议用户关闭ShuffleService功能，即在“Spark-defaults.conf”配置文件中将配置项“spark.shuffle.service.enabled”配置为“false”。

回答 在Spark中有个定期线程，通过连接RM监听AM的状态。由于连接RM超时，就会报上面的错误，且一直重试。RM中对重试次数有限制，默认是30次，每次间隔默认为30秒左右，每次重试时都会报上面的错误。超过次数后，driver才会退出。 RM中关于重试相关的配置项如表1所示。 表1 参数说明 参数 描述 默认值 yarn.resourcemanager.connect.max-wait.ms 连接RM的等待时间最大值。 900000 yarn.resourcemanager.connect.retry-interval.ms 重试连接RM的时间频率。 30000 重试次数=yarn.resourcemanager.connect.max-wait.ms/yarn.resourcemanager.connect.retry-interval.ms，即重试次数=连接RM的等待时间最大值/重试连接RM的时间频率。 在Spark客户端机器中，通过修改“conf/yarn-site.xml”文件，添加并配置“yarn.resourcemanager.connect.max-wait.ms”和“yarn.resourcemanager.connect.retry-interval.ms”，这样可以更改重试次数，Spark应用可以提早退出。

回答 在运行应用程序时，使用Executor参数“--executor-cores 4”，单进程中并行度高导致IO非常繁忙，以至于任务运行缓慢。 16/02/26 10:04:53 INFO TaskSetManager: Finished task 2139.0 in stage 1.0 (TID 151149) in 376455 ms on 10-196-115-2 (694/153378) 单个任务运行时间超过6分钟，从而导致连接超时问题，最终使得任务失败。 将参数中的核数设置为1，“--executor-cores 1”，任务正常完成，单个任务处理时间在合理范围之内(15秒左右)。 16/02/29 02:24:46 INFO TaskSetManager: Finished task 59564.0 in stage 1.0 (TID 208574) in 15088 ms on 10-196-115-6 (59515/153378) 因此，处理这类网络超时任务，可以减少单个Executor的核数来规避该类问题。

操作步骤 一个简单的流处理系统由以下三部分组件组成：数据源 + 接收器 + 处理器。数据源为Kafka，接受器为Streaming中的Kafka数据源接收器，处理器为Streaming。 对Streaming调优，就必须使该三个部件的性能都更优化。 数据源调优 在实际的应用场景中，数据源为了保证数据的容错性，会将数据保存在本地磁盘中，而Streaming的计算结果全部在内存中完成，数据源很有可能成为流式系统的最大瓶颈点。 对Kafka的性能调优，有以下几个点： 使用Kafka-0.8.2以后版本，可以使用异步模式的新Producer接口。 配置多个Broker的目录，设置多个IO线程，配置Topic合理的Partition个数。 详情请参见Kafka开源文档中的“性能调优”部分：http://kafka.apache.org/documentation.html。 接收器调优 Streaming中已有多种数据源的接收器，例如Kafka、Flume、MQTT、ZeroMQ等，其中Kafka的接收器类型最多，也是最成熟一套接收器。 Kafka包括三种模式的接收器API： KafkaReceiver：直接接收Kafka数据，进程异常后，可能出现数据丢失。 ReliableKafkaReceiver：通过ZooKeeper记录接收数据位移。 DirectKafka：直接通过RDD读取Kafka每个Partition中的数据，数据高可靠。 从实现上来看，DirectKafka的性能更优，实际测试上来看，DirectKafka也确实比其他两个API性能好了不少。因此推荐使用DirectKafka的API实现接收器。 数据接收器作为一个Kafka的消费者，对于它的配置优化，请参见Kafka开源文档：http://kafka.apache.org/documentation.html。 处理器调优 Spark Streaming的底层由Spark执行，因此大部分对于Spark的调优措施，都可以应用在Spark Streaming之中，例如： 数据序列化 配置内存 设置并行度 使用External Shuffle Service提升性能 在做Spark Streaming的性能优化时需注意一点，越追求性能上的优化，Spark Streaming整体的可靠性会越差。例如： “spark.streaming.receiver.writeAheadLog.enable”配置为“false”的时候，会明显减少磁盘的操作，提高性能，但由于缺少WAL机制，会出现异常恢复时，数据丢失。 因此，在调优Spark Streaming的时候，这些保证数据可靠性的配置项，在生产环境中是不能关闭的。 日志归档调优 参数“spark.eventLog.group.size”用来设置一个应用的JobHistory日志按照指定job个数分组，每个分组会单独创建一个文件记录日志，从而避免应用长期运行时形成单个过大日志造成JobHistory无法读取的问题，设置为“0”时表示不分组。 大部分Spark Streaming任务属于小型job，而且产生速度较快，会导致频繁的分组，产生大量日志小文件消耗磁盘I/O。建议增大此值，例如改为“1000”或更大值。