防止SparkListenerBus错误

Question

我在一个AWS EMR集群(EMR 5.0.0, Spark 2.0.0, 30 r3.4 × large)上运行Spark -submit应用程序。要启动脚本，我SSH到主节点，然后运行以下命令:

time spark-submit --conf spark.sql.shuffle.partitions=5000 
--conf spark.memory.storageFraction=0.3 --conf spark.memory.fraction=0.95 
--executor-memory 8G --driver-memory 10G dataframe_script.py

应用程序使用默认的AWS spark配置，其中有spark。Master =yarn, deploy-mode=client.

应用程序加载~220GB的数据，进行类似sql的聚合，然后写入s3。写入的数据看起来是经过正确处理的。当代码运行时，我看到一条错误消息，但代码继续运行:

ERROR LiveListenerBus: Dropping SparkListenerEvent because no remaining room in event queue. This likely means one of the SparkListeners is too slow and cannot keep up with the rate at which tasks are being started by the scheduler.

应用程序完成写入后，应用程序在>10分钟内没有返回命令行，并发出一个警告:

WARN ExecutorAllocationManager: No stages are running, but numRunningTasks != 0

然后是数万行错误信息:

16/10/12 00:40:03 ERROR LiveListenerBus: SparkListenerBus has already stopped! Dropping event SparkListenerExecutorMetricsUpdate(176,WrappedArray())

进度条也继续在错误消息之间移动，例如:

[Stage 17:=================================================>   (465 + 35) / 500]

主步骤的写入和结束代码:

def main():
    # some processing
    df.select(selection_list).write.json('s3path', compression=codec)
    print 'Done saving, shutting down'
    sc.stop()

之前有一个StackOverflow问题，它指的是这个JIRA。看起来有一个旧版本的Spark修复，但我不太明白问题是什么。

如何避免这些错误信息?

Answer 1

我想我找到问题了。在我的Spark脚本中，我在main()函数之外初始化SparkContext，但在main函数内部停止它。当脚本退出并试图第二次关闭SparkContext时，这会导致问题。通过将SparkContext初始化移到main函数中，大多数错误都消失了。