Package org.apache.spark.util

Interface LastAttemptAccumulator<IN,OUT,PARTIAL>

All Superinterfaces:
org.apache.spark.internal.Logging
public interface LastAttemptAccumulator<IN,OUT,PARTIAL> extends org.apache.spark.internal.Logging
A trait that can be mixed into a subclass of AccumulatorV2 to track the "logical" value of the "last attempt" of the execution using the accumulator - aggregated from the last attempts of any Task that calculated some RDD partitions and used this accumulator, and discarding any values coming from earlier attempts that have been recomputed. If the accumulator is used by multiple RDDs, the last attempt value is tracked separately for each, and can be retrieved for each or all of them separately, see lastAttemptValueForX methods. If the accumulator is used directly on the Spark Driver using AccumulatorV2.add(IN), that value is considered the last attempt value. If the accumulator was both used in Tasks and updated directly on the driver, it can't determine what should be considered the last attempt, and lastAttemptValueForX methods will return None.

Contract for driver-only updates: A driver-side value (set via AccumulatorV2.add(IN) on the driver, outside any Task) is only returned by methods that do not narrow by RDD, namely lastAttemptValueForAllRDDs() and lastAttemptValueForHighestRDDId(). Methods that narrow to specific RDDs or RDD scopes (lastAttemptValueForRDDId(int), lastAttemptValueForRDDIds(scala.collection.immutable.Seq<java.lang.Object>), lastAttemptValueForRDDScopes(scala.collection.immutable.Seq<java.lang.String>)) return the zero value when a driver-only value is present, because a driver-side update cannot be attributed to any particular RDD or scope.

LastAttemptAccumulator is not reset by the AccumulatorV2.reset() method implementation, and its state is not copied by the AccumulatorV2.copy() method implementation, and it should not be serialized to the Executors. The internal state should only be initialized by the initializeLastAttemptAccumulator(scala.reflect.ClassTag<PARTIAL>) method on the "main" instance of the accumulator, that was created and registered with AccumulatorContext with AccumulatorV2.register(org.apache.spark.SparkContext, scala.Option<java.lang.String>, boolean). All the interfaces of LastAttemptAccumulator: mergeLastAttempt(org.apache.spark.util.AccumulatorV2<?,?>,org.apache.spark.rdd.RDD<?>,org.apache.spark.scheduler.TaskInfo,int,int,java.util.Properties) (used only by DAGScheduler) and lastAttemptValueForX, logAccumulatorState() (used by the using code) should only be invoked on that instance, on the Spark Driver.

The LastAttemptAccumulator is not thread-safe. mergeLastAttempt(org.apache.spark.util.AccumulatorV2<?,?>,org.apache.spark.rdd.RDD<?>,org.apache.spark.scheduler.TaskInfo,int,int,java.util.Properties) should only be used by DAGScheduler, by the scheduler thread. Retrieving the value using lastAttemptValueForXXX while it is concurrently updated (execution is running) can produce some inconsistencies, but should not crash. If an RDD using the LastAttemptAccumulator is used concurrently by multiple actions that all try to recompute it, it may produce unexpected results and the semantics of what is "last attempt" becomes ambiguous. This should not be done in practice, and will likely result in more unexpected behaviours in Spark.

Implementations must implement partialMergeVal() and partialMerge(PARTIAL) methods operating on PARTIAL type. In regular AccumulatorV2 implementations, the AccumulatorV2 object itself holds the intermediate value of the accumulator, and AccumulatorV2.merge(org.apache.spark.util.AccumulatorV2<IN, OUT>) method is used to merge these objects together. LastAttemptAccumulator needs to keep track of partial values of every partition of every RDD that used the accumulator, and holding a full AccumulatorV2 object for each would have a high overhead. Therefore, an implementation should be able to return PARTIAL value from partialMergeVal() that represents an intermediate mergeable value, and a partialMerge(PARTIAL) method that can merge that value into the accumulator. Implementations must also implement an isMergeable(org.apache.spark.util.AccumulatorV2<?,?>) method that checks if the other AccumulatorV2 is of a compatible type to be merged with this using partialMergeVal(). In regular AccumulatorV2 implementations, this check is normally done inside the AccumulatorV2.merge(org.apache.spark.util.AccumulatorV2<IN, OUT>) method, which is not used here.

If an implementation is used to keep user data in the accumulator, it should override accumulatorStoresUserData() to return true, to ensure correct structured logging annotation. Otherwise it should override it to false.

  • Nested Class Summary

    Nested classes/interfaces inherited from interface org.apache.spark.internal.Logging

    org.apache.spark.internal.Logging.LogStringContext, org.apache.spark.internal.Logging.SparkShellLoggingFilter

  • Method Summary

    Modifier and Type
    Method
    Description
    long
    accumulatorId()
     
    boolean
    accumulatorStoresUserData()
    Accumulator subclasses where metric values can contain user data (for example, maximum of processed values, observable metrics) as opposed to system measurements (for example, count of processed rows) should return true to ensure correct structured logging annotation.
    void
    assertValid()
    Set of assertions that should always hold for a valid LastAttemptAccumulator.
    scala.Option<OUT>
    getDirectDriverValue()
    Visible for testing.
    scala.Option<Object>
    getHighestRDDId()
    Visible for testing
    int
    getNumRDDs()
    Visible for testing
    boolean
    getValid()
    Visible for testing
    void
    initializeLastAttemptAccumulator(scala.reflect.ClassTag<PARTIAL> ct)
     
    boolean
    isMergeable(AccumulatorV2<?,?> other)
    Check if the other accumulator is mergeable with this one.
    boolean
    lastAttemptAccumulatorInitialized()
     
    boolean
    lastAttemptAccumulatorInvalid()
     
    void
    lastAttemptValueAggregateInternal(int rddId, LastAttemptAccumulator<IN,OUT,PARTIAL> acc)
    Accumulates last attempt values from given RDD into an acc.
    scala.Option<OUT>
    lastAttemptValueForAllRDDs()
    Returns the last attempt value of this accumulator, aggregated from all RDDs that ever returned any values for it.
    scala.Option<OUT>
    lastAttemptValueForHighestRDDId()
    Returns the last attempt value of this accumulator, aggregated from the RDD with the highest id that ever returned any values for it.
    scala.Option<OUT>
    lastAttemptValueForRDDId(int rddId)
    Returns the last attempt value of this accumulator, aggregated from a specific RDD.
    scala.Option<OUT>
    lastAttemptValueForRDDIds(scala.collection.immutable.Seq<Object> rddIds)
    Returns the last attempt value of this accumulator, aggregated from a set of RDDs.
    scala.Option<OUT>
    lastAttemptValueForRDDScopes(scala.collection.immutable.Seq<String> rddScopeIds)
    Returns the last attempt value of this accumulator, aggregated from RDDs with given scope ids.
    org.apache.spark.internal.LogEntry
    logAccumulatorState()
    Log entry to log debug information about the internal state of the accumulator.
    org.apache.spark.internal.LogEntry
    logAccumulatorUpdate(scala.Option<org.apache.spark.util.AccumulatorPartialVal<PARTIAL>> newAccumPartialValue, scala.Option<org.apache.spark.util.AccumulatorPartialVal<PARTIAL>> oldAccumPartialValue)
     
    org.apache.spark.internal.LogKey
    logKeyAccumulatorState()
     
    void
    mergeLastAttempt(AccumulatorV2<?,?> other, RDD<?> rdd, TaskInfo taskInfo, int stageId, int stageAttemptId, Properties localProperties)
    It needs Task and Stage information to reason about the last attempt.
    void
    partialMerge(PARTIAL otherVal)
    Merge together partial values of PARTIAL type returned by partialMergeVal.
    PARTIAL
    partialMergeVal()
    Return intermediate value of PARTIAL type that can be merged together by partialMerge.
    void
    resetLastAttemptAccumulator()
    Reset the state of the last attempt accumulator, discarding all the past attempts, and making it valid again if it was invalidated.
    void
    setValueIfOnDriverSide(OUT value)
    Check if the value is set on the driver side, not from within a task.
    void
    unexpectedLastAttemptMetricOperation(boolean invalidate, String reason, scala.Option<Throwable> exception)
     
    void
    unexpectedLastAttemptMetricUpdate(boolean invalidate, String reason, scala.Option<Throwable> exception, scala.Option<org.apache.spark.util.AccumulatorPartialVal<PARTIAL>> newAccumPartialValue, scala.Option<org.apache.spark.util.AccumulatorPartialVal<PARTIAL>> oldAccumPartialValue)
     

    Methods inherited from interface org.apache.spark.internal.Logging

    initializeForcefully, initializeLogIfNecessary, initializeLogIfNecessary, initializeLogIfNecessary$default$2, isTraceEnabled, log, logBasedOnLevel, logDebug, logDebug, logDebug, logDebug, logError, logError, logError, logError, logInfo, logInfo, logInfo, logInfo, logName, LogStringContext, logTrace, logTrace, logTrace, logTrace, logWarning, logWarning, logWarning, logWarning, MDC, org$apache$spark$internal$Logging$$log_, org$apache$spark$internal$Logging$$log__$eq, withLogContext

  • Method Details

    • accumulatorId

      long accumulatorId()

    • accumulatorStoresUserData

      boolean accumulatorStoresUserData()
      Accumulator subclasses where metric values can contain user data (for example, maximum of processed values, observable metrics) as opposed to system measurements (for example, count of processed rows) should return true to ensure correct structured logging annotation.
      Returns:
      (undocumented)

    • assertValid

      void assertValid()
      Set of assertions that should always hold for a valid LastAttemptAccumulator.

    • getDirectDriverValue

      scala.Option<OUT> getDirectDriverValue()
      Visible for testing.

    • getHighestRDDId

      scala.Option<Object> getHighestRDDId()
      Visible for testing

    • getNumRDDs

      int getNumRDDs()
      Visible for testing

    • getValid

      boolean getValid()
      Visible for testing

    • initializeLastAttemptAccumulator

      void initializeLastAttemptAccumulator(scala.reflect.ClassTag<PARTIAL> ct)

    • isMergeable

      boolean isMergeable(AccumulatorV2<?,?> other)
      Check if the other accumulator is mergeable with this one.

    • lastAttemptAccumulatorInitialized

      boolean lastAttemptAccumulatorInitialized()

    • lastAttemptAccumulatorInvalid

      boolean lastAttemptAccumulatorInvalid()

    • lastAttemptValueAggregateInternal

      void lastAttemptValueAggregateInternal(int rddId, LastAttemptAccumulator<IN,OUT,PARTIAL> acc)
      Accumulates last attempt values from given RDD into an acc.

    • lastAttemptValueForAllRDDs

      scala.Option<OUT> lastAttemptValueForAllRDDs()
      Returns the last attempt value of this accumulator, aggregated from all RDDs that ever returned any values for it.

      If the metric was used directly on the driver, and was not used in any RDD execution, the driver value will be used instead.

      Should be used only on the Spark Driver, on the instance of LastAttemptAccumulator that was created and registered in AccumulatorContext by AccumulatorV2.register(org.apache.spark.SparkContext, scala.Option<java.lang.String>, boolean).

      Returns:
      None if the last attempt value cannot be established, Some(value) otherwise.

    • lastAttemptValueForHighestRDDId

      scala.Option<OUT> lastAttemptValueForHighestRDDId()
      Returns the last attempt value of this accumulator, aggregated from the RDD with the highest id that ever returned any values for it.

      If the metric was used directly on the driver, and was not used in any RDD execution, the driver value will be used instead.

      Should be used only on the Spark Driver, on the instance of LastAttemptAccumulator that was created and registered in AccumulatorContext by AccumulatorV2.register(org.apache.spark.SparkContext, scala.Option<java.lang.String>, boolean).

      Returns:
      None if the last attempt value cannot be established, Some(value) otherwise.

    • lastAttemptValueForRDDId

      scala.Option<OUT> lastAttemptValueForRDDId(int rddId)
      Returns the last attempt value of this accumulator, aggregated from a specific RDD.

      Should be used only on the Spark Driver, on the instance of LastAttemptAccumulator that was created and registered in AccumulatorContext by AccumulatorV2.register(org.apache.spark.SparkContext, scala.Option<java.lang.String>, boolean).

      Parameters:
      rddId - (undocumented)
      Returns:
      None if the last attempt value cannot be established, Some(value) otherwise.

    • lastAttemptValueForRDDIds

      scala.Option<OUT> lastAttemptValueForRDDIds(scala.collection.immutable.Seq<Object> rddIds)
      Returns the last attempt value of this accumulator, aggregated from a set of RDDs.

      Should be used only on the Spark Driver, on the instance of LastAttemptAccumulator that was created and registered in AccumulatorContext by AccumulatorV2.register(org.apache.spark.SparkContext, scala.Option<java.lang.String>, boolean).

      Parameters:
      rddIds - (undocumented)
      Returns:
      None if the last attempt value cannot be established, Some(value) otherwise.

    • lastAttemptValueForRDDScopes

      scala.Option<OUT> lastAttemptValueForRDDScopes(scala.collection.immutable.Seq<String> rddScopeIds)
      Returns the last attempt value of this accumulator, aggregated from RDDs with given scope ids.

      Should be used only on the Spark Driver, on the instance of LastAttemptAccumulator that was created and registered in AccumulatorContext by AccumulatorV2.register(org.apache.spark.SparkContext, scala.Option<java.lang.String>, boolean).

      Parameters:
      rddScopeIds - (undocumented)
      Returns:
      None if the last attempt value cannot be established, Some(value) otherwise.

    • logAccumulatorState

      org.apache.spark.internal.LogEntry logAccumulatorState()
      Log entry to log debug information about the internal state of the accumulator.

    • logAccumulatorUpdate

      org.apache.spark.internal.LogEntry logAccumulatorUpdate(scala.Option<org.apache.spark.util.AccumulatorPartialVal<PARTIAL>> newAccumPartialValue, scala.Option<org.apache.spark.util.AccumulatorPartialVal<PARTIAL>> oldAccumPartialValue)

    • logKeyAccumulatorState

      org.apache.spark.internal.LogKey logKeyAccumulatorState()

    • mergeLastAttempt

      void mergeLastAttempt(AccumulatorV2<?,?> other, RDD<?> rdd, TaskInfo taskInfo, int stageId, int stageAttemptId, Properties localProperties)
      It needs Task and Stage information to reason about the last attempt.

      Called from a single thread in DAGScheduler, no synchronization needed. Should be used only on the Spark Driver, on the instance of LastAttemptAccumulator that was created and registered in AccumulatorContext by AccumulatorV2.register(org.apache.spark.SparkContext, scala.Option<java.lang.String>, boolean).

      Parameters:
      other - (undocumented)
      rdd - (undocumented)
      taskInfo - (undocumented)
      stageId - (undocumented)
      stageAttemptId - (undocumented)
      localProperties - (undocumented)

    • partialMerge

      void partialMerge(PARTIAL otherVal)
      Merge together partial values of PARTIAL type returned by partialMergeVal.

    • partialMergeVal

      PARTIAL partialMergeVal()
      Return intermediate value of PARTIAL type that can be merged together by partialMerge.

    • resetLastAttemptAccumulator

      void resetLastAttemptAccumulator()
      Reset the state of the last attempt accumulator, discarding all the past attempts, and making it valid again if it was invalidated.

    • setValueIfOnDriverSide

      void setValueIfOnDriverSide(OUT value)
      Check if the value is set on the driver side, not from within a task. This must be called from add and set methods of any AccumulatorV2 subclass supporting last attempt metrics to set what the value of the metric is after the operation.
      Parameters:
      value - (undocumented)

    • unexpectedLastAttemptMetricOperation

      void unexpectedLastAttemptMetricOperation(boolean invalidate, String reason, scala.Option<Throwable> exception)

    • unexpectedLastAttemptMetricUpdate

      void unexpectedLastAttemptMetricUpdate(boolean invalidate, String reason, scala.Option<Throwable> exception, scala.Option<org.apache.spark.util.AccumulatorPartialVal<PARTIAL>> newAccumPartialValue, scala.Option<org.apache.spark.util.AccumulatorPartialVal<PARTIAL>> oldAccumPartialValue)