使用 Spark collectionAccumulator 时出现 ConcurrentModificationException答案

【问题标题】：ConcurrentModificationException when using Spark collectionAccumulator使用 Spark collectionAccumulator 时出现 ConcurrentModificationException
【发布时间】：2018-12-02 17:38:27
【问题描述】：

我正在尝试在 Azure HDInsight 按需群集上运行基于 Spark 的应用程序，并且看到记录了许多 SparkExceptions（由 ConcurrentModificationExceptions 引起）。当我启动本地 Spark 实例时，应用程序运行时没有这些错误。

我已经看到similar errors when using accumulators 的报告，我的代码确实使用了 CollectionAccumulator，但是我在使用它的任何地方都放置了同步块，这并没有什么区别。累加器相关的代码如下所示：

class MySparkClass(sc : SparkContext) {
    val myAccumulator = sc.collectionAccumulator[MyRecord]

    override def add(record: MyRecord) = {
        synchronized {
            myAccumulator.add(record)
        }
    }

    override def endOfBatch() = {
        synchronized {
            myAccumulator.value.asScala.foreach((record: MyRecord) => {
                processIt(record)
            })
        }
    }
}

异常不会导致应用程序失败，但是当调用 endOfBatch 并且代码尝试从累加器中读取值时，它是空的并且永远不会调用 processIt。

我们在 Spark 2.3.0 版中使用 HDInsight version 3.6

18/11/26 11:04:37 WARN Executor: Issue communicating with driver in heartbeater
org.apache.spark.SparkException: Exception thrown in awaitResult: 
    at org.apache.spark.util.ThreadUtils$.awaitResult(ThreadUtils.scala:205)
    at org.apache.spark.rpc.RpcTimeout.awaitResult(RpcTimeout.scala:75)
    at org.apache.spark.rpc.RpcEndpointRef.askSync(RpcEndpointRef.scala:92)
    at org.apache.spark.executor.Executor.org$apache$spark$executor$Executor$$reportHeartBeat(Executor.scala:785)
    at org.apache.spark.executor.Executor$$anon$2$$anonfun$run$1.apply$mcV$sp(Executor.scala:814)
    at org.apache.spark.executor.Executor$$anon$2$$anonfun$run$1.apply(Executor.scala:814)
    at org.apache.spark.executor.Executor$$anon$2$$anonfun$run$1.apply(Executor.scala:814)
    at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:1988)
    at org.apache.spark.executor.Executor$$anon$2.run(Executor.scala:814)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)
Caused by: java.util.ConcurrentModificationException
    at java.util.ArrayList.writeObject(ArrayList.java:770)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at java.io.ObjectStreamClass.invokeWriteObject(ObjectStreamClass.java:1140)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1496)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.defaultWriteObject(ObjectOutputStream.java:441)
    at java.util.Collections$SynchronizedCollection.writeObject(Collections.java:2081)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at java.io.ObjectStreamClass.invokeWriteObject(ObjectStreamClass.java:1140)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1496)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.writeArray(ObjectOutputStream.java:1378)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1174)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.writeArray(ObjectOutputStream.java:1378)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1174)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
    at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:43)
    at org.apache.spark.rpc.netty.RequestMessage.serialize(NettyRpcEnv.scala:565)
    at org.apache.spark.rpc.netty.NettyRpcEnv.ask(NettyRpcEnv.scala:231)
    at org.apache.spark.rpc.netty.NettyRpcEndpointRef.ask(NettyRpcEnv.scala:523)
    at org.apache.spark.rpc.RpcEndpointRef.askSync(RpcEndpointRef.scala:91)
    ... 13 more

以下代码是重现问题的更独立的示例。 MyRecord 是一个仅包含数值的简单案例类。代码在本地运行没有错误，但在 HDInsight 群集上会产生上述错误。

object MainDemo {
    def main(args: Array[String]) {
        val sparkContext = SparkSession.builder.master("local[4]").getOrCreate().sparkContext
        val myAccumulator = sparkContext.collectionAccumulator[MyRecord]

        sparkContext.binaryFiles("/my/files/here").foreach(_ => {
            for(i <- 1 to 100000) {
                val record = MyRecord(i, 0, 0)
                myAccumulator.add(record)
            }
        })

        myAccumulator.value.asScala.foreach((record: MyRecord) => {
            // we expect this to be called once for each record that we 'add' above,
            // but it is never called
            println(record)
        })
    }
}

【问题讨论】：

这听起来像是一个愚蠢的建议，但我并不完全相信这个错误来自您的累加器。所以有两个测试我想建议：1）用MySparkClass修改你的工作，所有（假）数据都预先添加到构造函数的累加器中，然后它add 从未修改过。 2) 使用修改为使用写时复制逻辑的代码进行测试运行：制作myAccumulator var 而不是val 并使用copy-add-assign 循环而不仅仅是add。这是 100% 线程安全的，但速度很慢。我敢打赌错误仍然存在。
链接到 Apache 网站上的错误报告：issues.apache.org/jira/browse/SPARK-26183

标签： scala azure apache-spark azure-hdinsight

【解决方案1】：

我怀疑同步块是否真的有帮助。 CustomeAccumulators 或所有其他累加器不是线程安全的。他们实际上不必这样做，因为 spark 驱动程序用于在任务完成（成功或失败）后更新累加器值的 DAGScheduler.updateAccumulators 方法仅在运行调度循环的单个线程上执行。除此之外，对于具有自己的本地累加器引用的工作人员来说，它们是只写数据结构，而只有驱动程序才允许访问累加器的值。当你说它在本地模式下工作，因为它是单个 JVM 但在集群模式下，它们是不同的 JVM 和 java 实例，正在触发 PRC 调用以启用通信。

你的 MyRecord 对象是什么样子的，如果你只是用 .value 结束你的行而不是在它上面有一个迭代器会有所帮助。试试吧。

myAccumulator.value

【讨论】：

add 方法是从 Executor 代码调用的，endOfBatch 是从驱动程序代码调用的。我不确定您在本地模式下要求什么，但是当我在本地 PC 上运行该应用程序时，它可以使用 local[4] 的主 URL - 如果您需要更多信息，请询问。
我需要遍历已添加到累加器中的值，如果我进行您建议的编辑，那么processIt 将如何被调用？ MyRecord 类是一个简单的案例类，具有各种数字和布尔字段
您能否向我们展示使用您的类的代码（或者至少是重现错误的最小示例）？
@Oli 我在问题中添加了一个示例
这很奇怪......我也设法在本地运行您的代码并获得了预期的行为。出于好奇，你为什么要使用累加器来收集这么多数据给司机？

【解决方案2】：

只有在 RDD 上的某些操作被调用（collect 或count）之后才读取累加器。

此外，您不需要在累加器上进行同步，因为每个分区都会为其分配一个独立的副本。

【讨论】：