Package org.apache.spark.sql.util

Class PartitionKeyedAccumulator<T>

Object
org.apache.spark.util.AccumulatorV2<scala.Tuple2<Object,T>,Map<Object,T>>
org.apache.spark.sql.util.PartitionKeyedAccumulator<T>
Type Parameters:
T - the per-partition value type. Must be non-null (ConcurrentHashMap forbids nulls).
All Implemented Interfaces:
Serializable
public class PartitionKeyedAccumulator<T> extends AccumulatorV2<scala.Tuple2<Object,T>,Map<Object,T>>
An AccumulatorV2 that records one value of type T per partition, keyed by partition id with LAST-WRITE-WINS merge. When the same partition is recorded more than once -- e.g. duplicate cross-executor computes, or speculative tasks -- the later value replaces the earlier one rather than aggregating, so each partition contributes exactly once. The key set is the set of recorded partitions, and callers fold the values (see <A>foldValues(A,scala.Function2<A,T,A>)) to derive de-duplicated aggregates; a plain summing accumulator would instead over-count under duplicate computes.

add is expected to be called once per task (e.g. from a task completion listener) with that partition's value, so a partition is recorded even when it produced nothing. Updates from failed/interrupted tasks are dropped by the accumulator framework (it is not countFailedValues), so only complete per-partition values are ever merged.

Backed by a ConcurrentHashMap, whose per-entry atomicity is sufficient here: add and the putAll in merge are last-write-wins per key, and the reads (value, accumulatedNumPartitions, foldValues) only require thread-safety and eventual consistency -- they are weakly consistent during concurrent updates but exact once all updates have been merged. This avoids any explicit locking (and the nested-lock pattern a two-map merge would otherwise need).

See Also:

  • Constructor Details

    • PartitionKeyedAccumulator

      public PartitionKeyedAccumulator()

  • Method Details

    • accumulatedNumPartitions

      public long accumulatedNumPartitions()
      Number of distinct partitions that have been recorded.

    • add

      public void add(scala.Tuple2<Object,T> v)
      Description copied from class: AccumulatorV2
      Takes the inputs and accumulates.
      Specified by:
      add in class AccumulatorV2<scala.Tuple2<Object,T>,Map<Object,T>>
      Parameters:
      v - (undocumented)

    • copy

      public PartitionKeyedAccumulator<T> copy()
      Description copied from class: AccumulatorV2
      Creates a new copy of this accumulator.
      Specified by:
      copy in class AccumulatorV2<scala.Tuple2<Object,T>,Map<Object,T>>
      Returns:
      (undocumented)

    • copyAndReset

      public PartitionKeyedAccumulator<T> copyAndReset()
      Description copied from class: AccumulatorV2
      Creates a new copy of this accumulator, which is zero value. i.e. call isZero on the copy must return true.
      Overrides:
      copyAndReset in class AccumulatorV2<scala.Tuple2<Object,T>,Map<Object,T>>
      Returns:
      (undocumented)

    • foldValues

      public <A> A foldValues(A zero, scala.Function2<A,T,A> op)
      Folds the per-partition values (each partition counted once) into a single aggregate.

    • isZero

      public boolean isZero()
      Description copied from class: AccumulatorV2
      Returns if this accumulator is zero value or not. e.g. for a counter accumulator, 0 is zero value; for a list accumulator, Nil is zero value.
      Specified by:
      isZero in class AccumulatorV2<scala.Tuple2<Object,T>,Map<Object,T>>
      Returns:
      (undocumented)

    • merge

      public void merge(AccumulatorV2<scala.Tuple2<Object,T>,Map<Object,T>> other)
      Description copied from class: AccumulatorV2
      Merges another same-type accumulator into this one and update its state, i.e. this should be merge-in-place.
      Specified by:
      merge in class AccumulatorV2<scala.Tuple2<Object,T>,Map<Object,T>>
      Parameters:
      other - (undocumented)

    • reset

      public void reset()
      Description copied from class: AccumulatorV2
      Resets this accumulator, which is zero value. i.e. call isZero must return true.
      Specified by:
      reset in class AccumulatorV2<scala.Tuple2<Object,T>,Map<Object,T>>

    • value

      public Map<Object,T> value()
      Description copied from class: AccumulatorV2
      Defines the current value of this accumulator
      Specified by:
      value in class AccumulatorV2<scala.Tuple2<Object,T>,Map<Object,T>>
      Returns:
      (undocumented)