Github user squito commented on a diff in the pull request: https://github.com/apache/spark/pull/11105#discussion_r56417617 --- Diff: core/src/main/scala/org/apache/spark/Accumulable.scala --- @@ -53,42 +54,83 @@ import org.apache.spark.util.Utils * for system and time metrics like serialization time or bytes spilled, * and false for things with absolute values like number of input rows. * This should be used for internal metrics only. - * @tparam R the full accumulated data (result type) + * @param consistent if this [[Accumulable]] is consistent. Consistent [[Accumulable]]s will only + * have values added once for each RDD/Partition/Shuffle combination. This + * prevents double counting on reevaluation. Partial evaluation of a partition + * will not increment a consistent [[Accumulable]]. Consistent [[Accumulable]]s + * are currently experimental and the behaviour may change in future versions. + * Consistent [[Accumulable]]s are designed for counting "data properties". + * @tparam R the full accumulated data * @tparam T partial data that can be added in */ -class Accumulable[R, T] private ( +class Accumulable[R, T] private[spark] ( val id: Long, // SI-8813: This must explicitly be a private val, or else scala 2.11 doesn't compile @transient private val initialValue: R, param: AccumulableParam[R, T], val name: Option[String], internal: Boolean, - private[spark] val countFailedValues: Boolean) + private[spark] val countFailedValues: Boolean, + private[spark] val consistent: Boolean) extends Serializable { private[spark] def this( initialValue: R, param: AccumulableParam[R, T], + internal: Boolean, + countFailedValues: Boolean, + consistent: Boolean) = { + this(Accumulators.newId(), initialValue, param, None, internal, countFailedValues, consistent) + } + + def this( + initialValue: R, + param: AccumulableParam[R, T], name: Option[String], internal: Boolean, countFailedValues: Boolean) = { - this(Accumulators.newId(), initialValue, param, name, internal, countFailedValues) + this(Accumulators.newId(), initialValue, param, name, internal, countFailedValues, + false /* consistent */) } - private[spark] def this( + def this( initialValue: R, param: AccumulableParam[R, T], name: Option[String], - internal: Boolean) = { - this(initialValue, param, name, internal, false /* countFailedValues */) + internal: Boolean, + countFailedValues: Boolean, + consistent: Boolean) = { + this(Accumulators.newId(), initialValue, param, name, internal, countFailedValues, consistent) } - def this(initialValue: R, param: AccumulableParam[R, T], name: Option[String]) = + def this( + initialValue: R, + param: AccumulableParam[R, T], + name: Option[String], + internal: Boolean) = + this(initialValue, param, name, internal, false /* countFailed */) + + def this( + initialValue: R, + param: AccumulableParam[R, T], + name: Option[String]) = this(initialValue, param, name, false /* internal */) - def this(initialValue: R, param: AccumulableParam[R, T]) = this(initialValue, param, None) + def this( + initialValue: R, + param: AccumulableParam[R, T]) = + this(initialValue, param, None) @volatile @transient private var value_ : R = initialValue // Current value on driver + + // For consistent accumulators pending and processed updates. + // Completed and processed are keyed by (rdd id, shuffle id, partition id) --- End diff -- typo: completed and pending
--- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. --- --------------------------------------------------------------------- To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org