Got an Accumulator error after adding some task metrics in Spark 2.0.2

[ShaneTian <ti...@ict.ac.cn>]

Hello all

I've tried to add some task metrics in
org.apache.spark.executor.ShuffleReadMetrics.scala in Spark 2.0.2, following
the format of other existing metrics, but when submitting applications, I
got these errors:

ERROR TaskSetManager: Failed to serialize task 0, not attempting to retry
it.
java.lang.UnsupportedOperationException: Accumulator must be registered
before send to executor
   at
org.apache.spark.util.AccumulatorV2.writeReplace(AccumulatorV2.scala:158)
   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.invokeWriteReplace(ObjectStreamClass.java:1118)
   at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1136)
   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.writeObject(ObjectOutputStream.java:348)
   at
org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:43)
   at
org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:100)
   at
org.apache.spark.scheduler.Task$.serializeWithDependencies(Task.scala:231)
   at
org.apache.spark.scheduler.TaskSetManager.resourceOffer(TaskSetManager.scala:458)
   ...
16/12/21 16:16:42 ERROR TaskSchedulerImpl: Resource offer failed, task set
TaskSet_0 was not serializable
Exception in thread "main" org.apache.spark.SparkException: Job aborted due
to stage failure: Failed to serialize task 0, not attempting to retry it.
Exception during serialization: java.lang.UnsupportedOperationException:
Accumulator must be registered before send to executor
   at
org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1454)
   at
org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1442)
   at
org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1441)
   at
scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
   at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
   at
org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1441)
   at
org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:811)
   at
org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:811)
   at scala.Option.foreach(Option.scala:257)
   at
org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:811)
   at
org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1667)
   at
org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1622)
   at
org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1611)
 ...
16/12/21 16:16:42 ERROR LiveListenerBus: SparkListenerBus has already
stopped! Dropping event    
SparkListenerBlockManagerAdded(1482308202401,BlockManagerId(7, 172.18.11.3,
42715),12409896960)
16/12/21 16:16:42 ERROR LiveListenerBus: SparkListenerBus has already
stopped! Dropping event    
SparkListenerBlockManagerAdded(1482308202445,BlockManagerId(5,
172.18.11.121, 41654),12409896960)

It seems like the Accumulator of task metrics has not been registered before
being used, but I also added the new metrics in the nameToAccums map in
TaskMetrics.scala:

private[spark] lazy val nameToAccums = LinkedHashMap(
  ...
  // add by txh
  shuffleRead.SHUFFLE_READ_TIME -> shuffleReadMetrics._shuffleReadTime,
  shuffleRead.SHUFFLE_MERGE_TIME -> shuffleReadMetrics._shuffleMergeTime,
  shuffleRead.SHUFFLE_MERGE_MEMORY ->
shuffleReadMetrics._shuffleMergeMemory,
  shuffleRead.SHUFFLE_USE_MEMORY -> shuffleReadMetrics._shuffleUseMemory,
  ...
)

What else should I add to make these new metrics be registered? Thanks very
much
ShuffleReadMetrics.scala
<http://apache-spark-developers-list.1001551.n3.nabble.com/file/n20330/ShuffleReadMetrics.scala>  



--
View this message in context: http://apache-spark-developers-list.1001551.n3.nabble.com/Got-an-Accumulator-error-after-adding-some-task-metrics-in-Spark-2-0-2-tp20330.html
Sent from the Apache Spark Developers List mailing list archive at Nabble.com.

---------------------------------------------------------------------
To unsubscribe e-mail: dev-unsubscribe@spark.apache.org