flink - accumulator

newJobStatus match {
  case JobStatus.FINISHED =>
  try {
    val accumulatorResults = executionGraph.getAccumulatorsSerialized()
    val result = new SerializedJobExecutionResult(
      jobID,
      jobInfo.duration,
      accumulatorResults)

    jobInfo.client ! decorateMessage(JobResultSuccess(result))
  }

public Map<String, Object> getAccumulators(JobID jobID, ClassLoader loader) throws Exception {
    ActorGateway jobManagerGateway = getJobManagerGateway();
    
    Future<Object> response;
    try {
        response = jobManagerGateway.ask(new RequestAccumulatorResults(jobID), timeout);
    } catch (Exception e) {
        throw new Exception("Failed to query the job manager gateway for accumulators.", e);
    }

case message: AccumulatorMessage => handleAccumulatorMessage(message)

private def handleAccumulatorMessage(message: AccumulatorMessage): Unit = {
message match {
  case RequestAccumulatorResults(jobID) =>
    try {
      currentJobs.get(jobID) match {
        case Some((graph, jobInfo)) =>
          val accumulatorValues = graph.getAccumulatorsSerialized()
          sender() ! decorateMessage(AccumulatorResultsFound(jobID, accumulatorValues))
        case None =>
          archive.forward(message)
      }
    }

/**
 * Gets a serialized accumulator map.
 * @return The accumulator map with serialized accumulator values.
 * @throws IOException
 */
public Map<String, SerializedValue<Object>> getAccumulatorsSerialized() throws IOException {

    Map<String, Accumulator<?, ?>> accumulatorMap = aggregateUserAccumulators();

    Map<String, SerializedValue<Object>> result = new HashMap<String, SerializedValue<Object>>();
    for (Map.Entry<String, Accumulator<?, ?>> entry : accumulatorMap.entrySet()) {
        result.put(entry.getKey(), new SerializedValue<Object>(entry.getValue().getLocalValue()));
    }

    return result;
}

/**
 * Merges all accumulator results from the tasks previously executed in the Executions.
 * @return The accumulator map
 */
public Map<String, Accumulator<?,?>> aggregateUserAccumulators() {

    Map<String, Accumulator<?, ?>> userAccumulators = new HashMap<String, Accumulator<?, ?>>();

    for (ExecutionVertex vertex : getAllExecutionVertices()) {
        Map<String, Accumulator<?, ?>> next = vertex.getCurrentExecutionAttempt().getUserAccumulators();
        if (next != null) {
            AccumulatorHelper.mergeInto(userAccumulators, next);
        }
    }

    return userAccumulators;
}

public static void mergeInto(Map<String, Accumulator<?, ?>> target, Map<String, Accumulator<?, ?>> toMerge) {
    for (Map.Entry<String, Accumulator<?, ?>> otherEntry : toMerge.entrySet()) {
        Accumulator<?, ?> ownAccumulator = target.get(otherEntry.getKey());
        if (ownAccumulator == null) {
            // Create initial counter (copy!)
            target.put(otherEntry.getKey(), otherEntry.getValue().clone());
        }
        else {
            // Both should have the same type
            AccumulatorHelper.compareAccumulatorTypes(otherEntry.getKey(),
                    ownAccumulator.getClass(), otherEntry.getValue().getClass());
            // Merge target counter with other counter
            mergeSingle(ownAccumulator, otherEntry.getValue());
        }
    }
}

case Heartbeat(instanceID, metricsReport, accumulators) =>
  updateAccumulators(accumulators)

private def updateAccumulators(accumulators : Seq[AccumulatorSnapshot]) = {
    accumulators foreach {
      case accumulatorEvent =>
        currentJobs.get(accumulatorEvent.getJobID) match {
          case Some((jobGraph, jobInfo)) =>
            future {
              jobGraph.updateAccumulators(accumulatorEvent)
            }(context.dispatcher)
          case None =>
          // ignore accumulator values for old job
        }
    }
}

/**
 * Updates the accumulators during the runtime of a job. Final accumulator results are transferred
 * through the UpdateTaskExecutionState message.
 * @param accumulatorSnapshot The serialized flink and user-defined accumulators
 */
public void updateAccumulators(AccumulatorSnapshot accumulatorSnapshot) {
    Map<AccumulatorRegistry.Metric, Accumulator<?, ?>> flinkAccumulators;
    Map<String, Accumulator<?, ?>> userAccumulators;
    try {
        flinkAccumulators = accumulatorSnapshot.deserializeFlinkAccumulators();
        userAccumulators = accumulatorSnapshot.deserializeUserAccumulators(userClassLoader);

        ExecutionAttemptID execID = accumulatorSnapshot.getExecutionAttemptID();
        Execution execution = currentExecutions.get(execID);
        if (execution != null) {
            execution.setAccumulators(flinkAccumulators, userAccumulators);
        }
    }
}

/**
 * Update accumulators (discarded when the Execution has already been terminated).
 * @param flinkAccumulators the flink internal accumulators
 * @param userAccumulators the user accumulators
 */
public void setAccumulators(Map<AccumulatorRegistry.Metric, Accumulator<?, ?>> flinkAccumulators,
                            Map<String, Accumulator<?, ?>> userAccumulators) {
    synchronized (accumulatorLock) {
        if (!state.isTerminal()) {
            this.flinkAccumulators = flinkAccumulators;
            this.userAccumulators = userAccumulators;
        }
    }
}

 /**
   * Sends a heartbeat message to the JobManager (if connected) with the current
   * metrics report.
   */
  protected def sendHeartbeatToJobManager(): Unit = {
    try {
      val metricsReport: Array[Byte] = metricRegistryMapper.writeValueAsBytes(metricRegistry)

      val accumulatorEvents =
        scala.collection.mutable.Buffer[AccumulatorSnapshot]()

      runningTasks foreach {
        case (execID, task) =>
          val registry = task.getAccumulatorRegistry
          val accumulators = registry.getSnapshot
          accumulatorEvents.append(accumulators)
      }

       currentJobManager foreach {
        jm => jm ! decorateMessage(Heartbeat(instanceID, metricsReport, accumulatorEvents))
      }
    }
  }

/**
 * Main accumulator registry which encapsulates internal and user-defined accumulators.
 */
public class AccumulatorRegistry {

    protected static final Logger LOG = LoggerFactory.getLogger(AccumulatorRegistry.class);

    protected final JobID jobID;  //accumulators所属的Job
    protected final ExecutionAttemptID taskID; //taskID

    /* Flink's internal Accumulator values stored for the executing task. */
    private final Map<Metric, Accumulator<?, ?>> flinkAccumulators =   //内部的Accumulators
            new HashMap<Metric, Accumulator<?, ?>>();

    /* User-defined Accumulator values stored for the executing task. */
    private final Map<String, Accumulator<?, ?>> userAccumulators = new HashMap<>(); //用户定义的Accumulators

    /* The reporter reference that is handed to the reporting tasks. */
    private final ReadWriteReporter reporter; 
    
    /**
     * Creates a snapshot of this accumulator registry.
     * @return a serialized accumulator map
     */
    public AccumulatorSnapshot getSnapshot() {
        try {
            return new AccumulatorSnapshot(jobID, taskID, flinkAccumulators, userAccumulators);
        } catch (IOException e) {
            LOG.warn("Failed to serialize accumulators for task.", e);
            return null;
        }
    }
}

public AccumulatorSnapshot(JobID jobID, ExecutionAttemptID executionAttemptID,
                        Map<AccumulatorRegistry.Metric, Accumulator<?, ?>> flinkAccumulators,
                        Map<String, Accumulator<?, ?>> userAccumulators) throws IOException {
    this.jobID = jobID;
    this.executionAttemptID = executionAttemptID;
    this.flinkAccumulators = new SerializedValue<Map<AccumulatorRegistry.Metric, Accumulator<?, ?>>>(flinkAccumulators);
    this.userAccumulators = new SerializedValue<Map<String, Accumulator<?, ?>>>(userAccumulators);
}

/**
 * Accumulator based reporter for keeping track of internal metrics (e.g. bytes and records in/out)
 */
private static class ReadWriteReporter implements Reporter {

    private LongCounter numRecordsIn = new LongCounter();
    private LongCounter numRecordsOut = new LongCounter();
    private LongCounter numBytesIn = new LongCounter();
    private LongCounter numBytesOut = new LongCounter();

    private ReadWriteReporter(Map<Metric, Accumulator<?,?>> accumulatorMap) {
        accumulatorMap.put(Metric.NUM_RECORDS_IN, numRecordsIn);
        accumulatorMap.put(Metric.NUM_RECORDS_OUT, numRecordsOut);
        accumulatorMap.put(Metric.NUM_BYTES_IN, numBytesIn);
        accumulatorMap.put(Metric.NUM_BYTES_OUT, numBytesOut);
    }

    @Override
    public void reportNumRecordsIn(long value) {
        numRecordsIn.add(value);
    }

    @Override
    public void reportNumRecordsOut(long value) {
        numRecordsOut.add(value);
    }

    @Override
    public void reportNumBytesIn(long value) {
        numBytesIn.add(value);
    }

    @Override
    public void reportNumBytesOut(long value) {
        numBytesOut.add(value);
    }
}

public DeserializationResult getNextRecord(T target) throws IOException {
    // check if we can get a full length;
    if (nonSpanningRemaining >= 4) {
        int len = this.nonSpanningWrapper.readInt();

        if (reporter != null) {
            reporter.reportNumBytesIn(len);
        }
        
        if (len <= nonSpanningRemaining - 4) {
            // we can get a full record from here
            target.read(this.nonSpanningWrapper);

            if (reporter != null) {
                reporter.reportNumRecordsIn(1);
            }

@Override
public SerializationResult addRecord(T record) throws IOException {
    int len = this.serializationBuffer.length();
    this.lengthBuffer.putInt(0, len);

    if (reporter != null) {
        reporter.reportNumBytesOut(len);
        reporter.reportNumRecordsOut(1);
    }