You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by mx...@apache.org on 2015/07/15 15:55:54 UTC
[1/4] flink git commit: [taskmanager] do not process message if not
connected to job manager
Repository: flink
Updated Branches:
refs/heads/master cbe3d9646 -> 8261ed543
[taskmanager] do not process message if not connected to job manager
This commit makes sure that task messages are only processed when
connected to a job manager. Otherwise, they are dropped. Before, the
message were processed in any case.
This closes #914.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/d592ee65
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/d592ee65
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/d592ee65
Branch: refs/heads/master
Commit: d592ee65f040bb0fa737cd1ac7fffec39ed9068f
Parents: cbe3d96
Author: Maximilian Michels <mx...@apache.org>
Authored: Wed Jul 15 12:16:24 2015 +0200
Committer: Maximilian Michels <mx...@apache.org>
Committed: Wed Jul 15 15:00:54 2015 +0200
----------------------------------------------------------------------
.../flink/runtime/taskmanager/TaskManager.scala | 160 +++++++++----------
1 file changed, 80 insertions(+), 80 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/d592ee65/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
index 44a0b04..1a35d01 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
@@ -307,99 +307,99 @@ extends Actor with ActorLogMessages with ActorSynchronousLogging {
if (!isConnected) {
log.debug(s"Dropping message $message because the TaskManager is currently " +
"not connected to a JobManager.")
- }
+ } else {
+ // we order the messages by frequency, to make sure the code paths for matching
+ // are as short as possible
+ message match {
+
+ // tell the task about the availability of a new input partition
+ case UpdateTaskSinglePartitionInfo(executionID, resultID, partitionInfo) =>
+ updateTaskInputPartitions(executionID, List((resultID, partitionInfo)))
+
+ // tell the task about the availability of some new input partitions
+ case UpdateTaskMultiplePartitionInfos(executionID, partitionInfos) =>
+ updateTaskInputPartitions(executionID, partitionInfos)
+
+ // discards intermediate result partitions of a task execution on this TaskManager
+ case FailIntermediateResultPartitions(executionID) =>
+ log.info("Discarding the results produced by task execution " + executionID)
+ if (network.isAssociated) {
+ try {
+ network.getPartitionManager.releasePartitionsProducedBy(executionID)
+ } catch {
+ case t: Throwable => killTaskManagerFatal(
+ "Fatal leak: Unable to release intermediate result partition data", t)
+ }
+ }
- // we order the messages by frequency, to make sure the code paths for matching
- // are as short as possible
- message match {
+ // notifies the TaskManager that the state of a task has changed.
+ // the TaskManager informs the JobManager and cleans up in case the transition
+ // was into a terminal state, or in case the JobManager cannot be informed of the
+ // state transition
- // tell the task about the availability of a new input partition
- case UpdateTaskSinglePartitionInfo(executionID, resultID, partitionInfo) =>
- updateTaskInputPartitions(executionID, List((resultID, partitionInfo)))
+ case updateMsg@UpdateTaskExecutionState(taskExecutionState: TaskExecutionState) =>
- // tell the task about the availability of some new input partitions
- case UpdateTaskMultiplePartitionInfos(executionID, partitionInfos) =>
- updateTaskInputPartitions(executionID, partitionInfos)
+ // we receive these from our tasks and forward them to the JobManager
+ currentJobManager foreach {
+ jobManager => {
+ val futureResponse = (jobManager ? updateMsg)(askTimeout)
- // discards intermediate result partitions of a task execution on this TaskManager
- case FailIntermediateResultPartitions(executionID) =>
- log.info("Discarding the results produced by task execution " + executionID)
- if (network.isAssociated) {
- try {
- network.getPartitionManager.releasePartitionsProducedBy(executionID)
- } catch {
- case t: Throwable => killTaskManagerFatal(
- "Fatal leak: Unable to release intermediate result partition data", t)
- }
- }
+ val executionID = taskExecutionState.getID
- // notifies the TaskManager that the state of a task has changed.
- // the TaskManager informs the JobManager and cleans up in case the transition
- // was into a terminal state, or in case the JobManager cannot be informed of the
- // state transition
+ futureResponse.mapTo[Boolean].onComplete {
+ // IMPORTANT: In the future callback, we cannot directly modify state
+ // but only send messages to the TaskManager to do those changes
+ case Success(result) =>
+ if (!result) {
+ self ! FailTask(executionID,
+ new Exception("Task has been cancelled on the JobManager."))
+ }
- case updateMsg @ UpdateTaskExecutionState(taskExecutionState: TaskExecutionState) =>
-
- // we receive these from our tasks and forward them to the JobManager
- currentJobManager foreach {
- jobManager => {
- val futureResponse = (jobManager ? updateMsg)(askTimeout)
-
- val executionID = taskExecutionState.getID
-
- futureResponse.mapTo[Boolean].onComplete {
- // IMPORTANT: In the future callback, we cannot directly modify state
- // but only send messages to the TaskManager to do those changes
- case Success(result) =>
- if (!result) {
- self ! FailTask(executionID,
- new Exception("Task has been cancelled on the JobManager."))
- }
-
- case Failure(t) =>
- self ! FailTask(executionID, new Exception(
- "Failed to send ExecutionStateChange notification to JobManager"))
- }(context.dispatcher)
+ case Failure(t) =>
+ self ! FailTask(executionID, new Exception(
+ "Failed to send ExecutionStateChange notification to JobManager"))
+ }(context.dispatcher)
+ }
}
- }
- // removes the task from the TaskManager and frees all its resources
- case TaskInFinalState(executionID) =>
- unregisterTaskAndNotifyFinalState(executionID)
+ // removes the task from the TaskManager and frees all its resources
+ case TaskInFinalState(executionID) =>
+ unregisterTaskAndNotifyFinalState(executionID)
- // starts a new task on the TaskManager
- case SubmitTask(tdd) =>
- submitTask(tdd)
+ // starts a new task on the TaskManager
+ case SubmitTask(tdd) =>
+ submitTask(tdd)
- // marks a task as failed for an external reason
- // external reasons are reasons other than the task code itself throwing an exception
- case FailTask(executionID, cause) =>
- val task = runningTasks.get(executionID)
- if (task != null) {
- task.failExternally(cause)
- } else {
- log.debug(s"Cannot find task to fail for execution ${executionID})")
- }
+ // marks a task as failed for an external reason
+ // external reasons are reasons other than the task code itself throwing an exception
+ case FailTask(executionID, cause) =>
+ val task = runningTasks.get(executionID)
+ if (task != null) {
+ task.failExternally(cause)
+ } else {
+ log.debug(s"Cannot find task to fail for execution ${executionID})")
+ }
- // cancels a task
- case CancelTask(executionID) =>
- val task = runningTasks.get(executionID)
- if (task != null) {
- task.cancelExecution()
- sender ! new TaskOperationResult(executionID, true)
- } else {
- log.debug(s"Cannot find task to cancel for execution ${executionID})")
- sender ! new TaskOperationResult(executionID, false,
+ // cancels a task
+ case CancelTask(executionID) =>
+ val task = runningTasks.get(executionID)
+ if (task != null) {
+ task.cancelExecution()
+ sender ! new TaskOperationResult(executionID, true)
+ } else {
+ log.debug(s"Cannot find task to cancel for execution ${executionID})")
+ sender ! new TaskOperationResult(executionID, false,
"No task with that execution ID was found.")
- }
+ }
- case PartitionState(taskExecutionId, taskResultId, partitionId, state) =>
- Option(runningTasks.get(taskExecutionId)) match {
- case Some(task) =>
- task.onPartitionStateUpdate(taskResultId, partitionId, state)
- case None =>
- log.debug(s"Cannot find task $taskExecutionId to respond with partition state.")
- }
+ case PartitionState(taskExecutionId, taskResultId, partitionId, state) =>
+ Option(runningTasks.get(taskExecutionId)) match {
+ case Some(task) =>
+ task.onPartitionStateUpdate(taskResultId, partitionId, state)
+ case None =>
+ log.debug(s"Cannot find task $taskExecutionId to respond with partition state.")
+ }
+ }
}
}
[4/4] flink git commit: [FLINK-2292][FLINK-1573] add live per-task
accumulators
Posted by mx...@apache.org.
[FLINK-2292][FLINK-1573] add live per-task accumulators
This refactors the accumulators to accumulate per task execution. The
accumulators are reported from the task managers periodically to the job
manager via the Heartbeat message. If the execution contains chained
tasks, the accumulators are chained as well. The final accumulator
results are reported via the UpdateTaskExecutionState message.
The accumulators are now saved in the Execution within the
ExecutionGraph. This makes the AccumulatorManager obsolete. It has been
removed for now. In the future, we might introduce some caching for the
web frontend visualization.
Two types of accumulators are available:
- external (user-defined via the RuntimeContext)
- internal (flink metrics defined in the invocables)
The internal (built-in) metrics are targeted at users who want to
monitor their programs, e.g. through the job manager's web frontend.
This closes #896.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/8261ed54
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/8261ed54
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/8261ed54
Branch: refs/heads/master
Commit: 8261ed5438278331c9cd760463273ad94d4bd410
Parents: d592ee6
Author: Maximilian Michels <mx...@apache.org>
Authored: Wed Jul 8 09:23:42 2015 +0200
Committer: Maximilian Michels <mx...@apache.org>
Committed: Wed Jul 15 15:19:06 2015 +0200
----------------------------------------------------------------------
.../api/common/accumulators/Accumulator.java | 1 -
.../api/common/functions/RuntimeContext.java | 10 +-
.../util/AbstractRuntimeUDFContext.java | 26 +-
.../functions/util/RuntimeUDFContext.java | 11 +-
.../common/operators/CollectionExecutor.java | 19 +-
.../functions/util/RuntimeUDFContextTest.java | 12 +-
.../base/FlatMapOperatorCollectionTest.java | 4 +-
.../operators/base/JoinOperatorBaseTest.java | 7 +-
.../common/operators/base/MapOperatorTest.java | 7 +-
.../base/PartitionMapOperatorTest.java | 6 +-
.../examples/java/wordcount/WordCount.java | 2 +-
.../java/org/apache/flink/api/java/DataSet.java | 18 +-
.../java/org/apache/flink/api/java/Utils.java | 7 +-
.../base/CoGroupOperatorCollectionTest.java | 5 +-
.../operators/base/GroupReduceOperatorTest.java | 6 +-
.../operators/base/JoinOperatorBaseTest.java | 6 +-
.../operators/base/ReduceOperatorTest.java | 6 +-
.../org/apache/flink/optimizer/Optimizer.java | 2 +-
.../runtime/accumulators/AccumulatorEvent.java | 49 ---
.../accumulators/AccumulatorRegistry.java | 147 +++++++++
.../accumulators/AccumulatorSnapshot.java | 84 ++++++
.../flink/runtime/execution/Environment.java | 9 +-
.../librarycache/BlobLibraryCacheManager.java | 2 +-
.../flink/runtime/executiongraph/Execution.java | 43 +++
.../runtime/executiongraph/ExecutionGraph.java | 100 ++++++-
.../io/network/api/reader/AbstractReader.java | 1 +
.../api/reader/AbstractRecordReader.java | 14 +-
.../io/network/api/reader/BufferReader.java | 6 +
.../io/network/api/reader/ReaderBase.java | 6 +
.../io/network/api/reader/RecordReader.java | 1 +
.../AdaptiveSpanningRecordDeserializer.java | 31 ++
.../api/serialization/RecordDeserializer.java | 6 +
.../api/serialization/RecordSerializer.java | 6 +
.../serialization/SpanningRecordSerializer.java | 16 +-
...llingAdaptiveSpanningRecordDeserializer.java | 34 ++-
.../io/network/api/writer/RecordWriter.java | 28 ++
.../task/AbstractIterativePactTask.java | 19 +-
.../iterative/task/IterationHeadPactTask.java | 4 +-
.../jobgraph/tasks/AbstractInvokable.java | 2 +-
.../accumulators/AccumulatorManager.java | 144 ---------
.../accumulators/JobAccumulators.java | 42 ---
.../flink/runtime/operators/DataSinkTask.java | 6 +
.../flink/runtime/operators/DataSourceTask.java | 27 +-
.../flink/runtime/operators/PactDriver.java | 2 +-
.../runtime/operators/PactTaskContext.java | 2 +-
.../runtime/operators/RegularPactTask.java | 133 +++------
.../operators/chaining/ChainedDriver.java | 14 +-
.../util/DistributedRuntimeUDFContext.java | 11 +-
.../runtime/taskmanager/RuntimeEnvironment.java | 32 +-
.../apache/flink/runtime/taskmanager/Task.java | 42 ++-
.../runtime/taskmanager/TaskExecutionState.java | 43 ++-
.../flink/runtime/util/SerializedValue.java | 2 +-
.../flink/runtime/jobmanager/JobManager.scala | 118 ++++----
.../runtime/messages/TaskManagerMessages.scala | 5 +-
.../accumulators/AccumulatorMessages.scala | 15 +-
.../flink/runtime/taskmanager/TaskManager.scala | 61 ++--
.../network/api/reader/AbstractReaderTest.java | 7 +
.../operators/testutils/DriverTestBase.java | 6 +-
.../operators/testutils/MockEnvironment.java | 9 +-
.../flink/runtime/taskmanager/TaskTest.java | 1 +
.../runtime/testingUtils/TestingCluster.scala | 2 -
.../testingUtils/TestingJobManager.scala | 10 +
.../TestingJobManagerMessages.scala | 10 +-
.../streaming/runtime/io/CoRecordReader.java | 11 +
.../io/StreamingAbstractRecordReader.java | 13 +-
.../io/StreamingMutableRecordReader.java | 1 +
.../runtime/tasks/OneInputStreamTask.java | 6 +
.../streaming/runtime/tasks/OutputHandler.java | 32 +-
.../runtime/tasks/StreamIterationHead.java | 9 +-
.../streaming/runtime/tasks/StreamTask.java | 16 +-
.../runtime/tasks/StreamingRuntimeContext.java | 5 +-
.../api/state/StatefulOperatorTest.java | 4 +-
.../streaming/runtime/io/BarrierBufferTest.java | 6 +
.../runtime/tasks/StreamMockEnvironment.java | 9 +-
.../flink/streaming/util/MockCoContext.java | 4 +-
.../flink/streaming/util/MockContext.java | 4 +-
.../streaming/util/SourceFunctionUtil.java | 4 +-
.../flink/tez/runtime/RegularProcessor.java | 4 +-
.../flink/test/util/RecordAPITestBase.java | 2 +-
.../test/util/ForkableFlinkMiniCluster.scala | 2 -
.../test/accumulators/AccumulatorITCase.java | 3 +-
.../AccumulatorIterativeITCase.java | 18 +-
.../accumulators/AccumulatorLiveITCase.java | 299 +++++++++++++++++++
.../apache/flink/yarn/ApplicationMaster.scala | 2 -
84 files changed, 1362 insertions(+), 599 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-core/src/main/java/org/apache/flink/api/common/accumulators/Accumulator.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/accumulators/Accumulator.java b/flink-core/src/main/java/org/apache/flink/api/common/accumulators/Accumulator.java
index 123d956..e49cc04 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/accumulators/Accumulator.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/accumulators/Accumulator.java
@@ -40,7 +40,6 @@ import java.io.Serializable;
* client
*/
public interface Accumulator<V, R extends Serializable> extends Serializable, Cloneable {
-
/**
* @param value
* The value to add to the accumulator object
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-core/src/main/java/org/apache/flink/api/common/functions/RuntimeContext.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/RuntimeContext.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/RuntimeContext.java
index eb84d1c..fb9d842 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/RuntimeContext.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/functions/RuntimeContext.java
@@ -20,8 +20,8 @@ package org.apache.flink.api.common.functions;
import java.io.IOException;
import java.io.Serializable;
-import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.accumulators.Accumulator;
@@ -100,10 +100,12 @@ public interface RuntimeContext {
<V, A extends Serializable> Accumulator<V, A> getAccumulator(String name);
/**
- * For system internal usage only. Use getAccumulator(...) to obtain a
- * accumulator. Use this as read-only.
+ * Returns a map of all registered accumulators for this task.
+ * The returned map must not be modified.
+ * @deprecated Use getAccumulator(..) to obtain the value of an accumulator.
*/
- HashMap<String, Accumulator<?, ?>> getAllAccumulators();
+ @Deprecated
+ Map<String, Accumulator<?, ?>> getAllAccumulators();
/**
* Convenience function to create a counter object for integers.
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-core/src/main/java/org/apache/flink/api/common/functions/util/AbstractRuntimeUDFContext.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/util/AbstractRuntimeUDFContext.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/util/AbstractRuntimeUDFContext.java
index f48eb57..13d79e7 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/util/AbstractRuntimeUDFContext.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/functions/util/AbstractRuntimeUDFContext.java
@@ -21,10 +21,10 @@ package org.apache.flink.api.common.functions.util;
import java.io.IOException;
import java.io.Serializable;
import java.util.Collections;
-import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Future;
+import com.google.common.base.Preconditions;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.accumulators.AccumulatorHelper;
@@ -53,32 +53,34 @@ public abstract class AbstractRuntimeUDFContext implements RuntimeContext {
private final ExecutionConfig executionConfig;
- private final HashMap<String, Accumulator<?, ?>> accumulators = new HashMap<String, Accumulator<?, ?>>();
-
+ private final Map<String, Accumulator<?, ?>> accumulators;
+
private final DistributedCache distributedCache;
-
-
+
+
public AbstractRuntimeUDFContext(String name,
int numParallelSubtasks, int subtaskIndex,
ClassLoader userCodeClassLoader,
- ExecutionConfig executionConfig)
+ ExecutionConfig executionConfig,
+ Map<String, Accumulator<?,?>> accumulators)
{
this(name, numParallelSubtasks, subtaskIndex, userCodeClassLoader, executionConfig,
- Collections.<String, Future<Path>>emptyMap());
+ accumulators, Collections.<String, Future<Path>>emptyMap());
}
-
+
public AbstractRuntimeUDFContext(String name,
int numParallelSubtasks, int subtaskIndex,
ClassLoader userCodeClassLoader,
ExecutionConfig executionConfig,
- Map<String, Future<Path>> cpTasks)
- {
+ Map<String, Accumulator<?,?>> accumulators,
+ Map<String, Future<Path>> cpTasks) {
this.name = name;
this.numParallelSubtasks = numParallelSubtasks;
this.subtaskIndex = subtaskIndex;
this.userCodeClassLoader = userCodeClassLoader;
this.executionConfig = executionConfig;
this.distributedCache = new DistributedCache(cpTasks);
+ this.accumulators = Preconditions.checkNotNull(accumulators);
}
@Override
@@ -137,8 +139,8 @@ public abstract class AbstractRuntimeUDFContext implements RuntimeContext {
}
@Override
- public HashMap<String, Accumulator<?, ?>> getAllAccumulators() {
- return this.accumulators;
+ public Map<String, Accumulator<?, ?>> getAllAccumulators() {
+ return Collections.unmodifiableMap(this.accumulators);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-core/src/main/java/org/apache/flink/api/common/functions/util/RuntimeUDFContext.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/functions/util/RuntimeUDFContext.java b/flink-core/src/main/java/org/apache/flink/api/common/functions/util/RuntimeUDFContext.java
index d116d00..1689138 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/functions/util/RuntimeUDFContext.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/functions/util/RuntimeUDFContext.java
@@ -24,6 +24,7 @@ import java.util.Map;
import java.util.concurrent.Future;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.functions.BroadcastVariableInitializer;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.core.fs.Path;
@@ -38,12 +39,14 @@ public class RuntimeUDFContext extends AbstractRuntimeUDFContext {
private final HashMap<String, List<?>> uninitializedBroadcastVars = new HashMap<String, List<?>>();
- public RuntimeUDFContext(String name, int numParallelSubtasks, int subtaskIndex, ClassLoader userCodeClassLoader, ExecutionConfig executionConfig) {
- super(name, numParallelSubtasks, subtaskIndex, userCodeClassLoader, executionConfig);
+ public RuntimeUDFContext(String name, int numParallelSubtasks, int subtaskIndex, ClassLoader userCodeClassLoader,
+ ExecutionConfig executionConfig, Map<String, Accumulator<?,?>> accumulators) {
+ super(name, numParallelSubtasks, subtaskIndex, userCodeClassLoader, executionConfig, accumulators);
}
- public RuntimeUDFContext(String name, int numParallelSubtasks, int subtaskIndex, ClassLoader userCodeClassLoader, ExecutionConfig executionConfig, Map<String, Future<Path>> cpTasks) {
- super(name, numParallelSubtasks, subtaskIndex, userCodeClassLoader, executionConfig, cpTasks);
+ public RuntimeUDFContext(String name, int numParallelSubtasks, int subtaskIndex, ClassLoader userCodeClassLoader,
+ ExecutionConfig executionConfig, Map<String, Future<Path>> cpTasks, Map<String, Accumulator<?,?>> accumulators) {
+ super(name, numParallelSubtasks, subtaskIndex, userCodeClassLoader, executionConfig, accumulators, cpTasks);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java b/flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java
index f605113..0a9146c 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java
@@ -184,8 +184,8 @@ public class CollectionExecutor {
// build the runtime context and compute broadcast variables, if necessary
RuntimeUDFContext ctx;
if (RichFunction.class.isAssignableFrom(typedOp.getUserCodeWrapper().getUserCodeClass())) {
- ctx = superStep == 0 ? new RuntimeUDFContext(operator.getName(), 1, 0, getClass().getClassLoader(), executionConfig) :
- new IterationRuntimeUDFContext(operator.getName(), 1, 0, classLoader, executionConfig);
+ ctx = superStep == 0 ? new RuntimeUDFContext(operator.getName(), 1, 0, getClass().getClassLoader(), executionConfig, accumulators) :
+ new IterationRuntimeUDFContext(operator.getName(), 1, 0, classLoader, executionConfig, accumulators);
for (Map.Entry<String, Operator<?>> bcInputs : operator.getBroadcastInputs().entrySet()) {
List<?> bcData = execute(bcInputs.getValue());
@@ -197,9 +197,6 @@ public class CollectionExecutor {
List<OUT> result = typedOp.executeOnCollections(inputData, ctx, executionConfig);
- if (ctx != null) {
- AccumulatorHelper.mergeInto(this.accumulators, ctx.getAllAccumulators());
- }
return result;
}
@@ -226,8 +223,8 @@ public class CollectionExecutor {
// build the runtime context and compute broadcast variables, if necessary
RuntimeUDFContext ctx;
if (RichFunction.class.isAssignableFrom(typedOp.getUserCodeWrapper().getUserCodeClass())) {
- ctx = superStep == 0 ? new RuntimeUDFContext(operator.getName(), 1, 0, classLoader, executionConfig) :
- new IterationRuntimeUDFContext(operator.getName(), 1, 0, classLoader, executionConfig);
+ ctx = superStep == 0 ? new RuntimeUDFContext(operator.getName(), 1, 0, classLoader, executionConfig, accumulators) :
+ new IterationRuntimeUDFContext(operator.getName(), 1, 0, classLoader, executionConfig, accumulators);
for (Map.Entry<String, Operator<?>> bcInputs : operator.getBroadcastInputs().entrySet()) {
List<?> bcData = execute(bcInputs.getValue());
@@ -239,9 +236,6 @@ public class CollectionExecutor {
List<OUT> result = typedOp.executeOnCollections(inputData1, inputData2, ctx, executionConfig);
- if (ctx != null) {
- AccumulatorHelper.mergeInto(this.accumulators, ctx.getAllAccumulators());
- }
return result;
}
@@ -485,8 +479,9 @@ public class CollectionExecutor {
private class IterationRuntimeUDFContext extends RuntimeUDFContext implements IterationRuntimeContext {
- public IterationRuntimeUDFContext(String name, int numParallelSubtasks, int subtaskIndex, ClassLoader classloader, ExecutionConfig executionConfig) {
- super(name, numParallelSubtasks, subtaskIndex, classloader, executionConfig);
+ public IterationRuntimeUDFContext(String name, int numParallelSubtasks, int subtaskIndex, ClassLoader classloader,
+ ExecutionConfig executionConfig, Map<String, Accumulator<?,?>> accumulators) {
+ super(name, numParallelSubtasks, subtaskIndex, classloader, executionConfig, accumulators);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-core/src/test/java/org/apache/flink/api/common/functions/util/RuntimeUDFContextTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/functions/util/RuntimeUDFContextTest.java b/flink-core/src/test/java/org/apache/flink/api/common/functions/util/RuntimeUDFContextTest.java
index c77a1b6..9189d5b 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/functions/util/RuntimeUDFContextTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/functions/util/RuntimeUDFContextTest.java
@@ -22,9 +22,11 @@ import static org.junit.Assert.*;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.HashMap;
import java.util.List;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.functions.BroadcastVariableInitializer;
import org.junit.Test;
@@ -34,7 +36,7 @@ public class RuntimeUDFContextTest {
@Test
public void testBroadcastVariableNotFound() {
try {
- RuntimeUDFContext ctx = new RuntimeUDFContext("test name", 3, 1, getClass().getClassLoader(), new ExecutionConfig());
+ RuntimeUDFContext ctx = new RuntimeUDFContext("test name", 3, 1, getClass().getClassLoader(), new ExecutionConfig(), new HashMap<String, Accumulator<?, ?>>());
try {
ctx.getBroadcastVariable("some name");
@@ -64,7 +66,7 @@ public class RuntimeUDFContextTest {
@Test
public void testBroadcastVariableSimple() {
try {
- RuntimeUDFContext ctx = new RuntimeUDFContext("test name", 3, 1, getClass().getClassLoader(), new ExecutionConfig());
+ RuntimeUDFContext ctx = new RuntimeUDFContext("test name", 3, 1, getClass().getClassLoader(), new ExecutionConfig(), new HashMap<String, Accumulator<?, ?>>());
ctx.setBroadcastVariable("name1", Arrays.asList(1, 2, 3, 4));
ctx.setBroadcastVariable("name2", Arrays.asList(1.0, 2.0, 3.0, 4.0));
@@ -98,7 +100,7 @@ public class RuntimeUDFContextTest {
@Test
public void testBroadcastVariableWithInitializer() {
try {
- RuntimeUDFContext ctx = new RuntimeUDFContext("test name", 3, 1, getClass().getClassLoader(), new ExecutionConfig());
+ RuntimeUDFContext ctx = new RuntimeUDFContext("test name", 3, 1, getClass().getClassLoader(), new ExecutionConfig(), new HashMap<String, Accumulator<?, ?>>());
ctx.setBroadcastVariable("name", Arrays.asList(1, 2, 3, 4));
@@ -123,7 +125,7 @@ public class RuntimeUDFContextTest {
@Test
public void testResetBroadcastVariableWithInitializer() {
try {
- RuntimeUDFContext ctx = new RuntimeUDFContext("test name", 3, 1, getClass().getClassLoader(), new ExecutionConfig());
+ RuntimeUDFContext ctx = new RuntimeUDFContext("test name", 3, 1, getClass().getClassLoader(), new ExecutionConfig(), new HashMap<String, Accumulator<?, ?>>());
ctx.setBroadcastVariable("name", Arrays.asList(1, 2, 3, 4));
@@ -146,7 +148,7 @@ public class RuntimeUDFContextTest {
@Test
public void testBroadcastVariableWithInitializerAndMismatch() {
try {
- RuntimeUDFContext ctx = new RuntimeUDFContext("test name", 3, 1, getClass().getClassLoader(), new ExecutionConfig());
+ RuntimeUDFContext ctx = new RuntimeUDFContext("test name", 3, 1, getClass().getClassLoader(), new ExecutionConfig(), new HashMap<String, Accumulator<?, ?>>());
ctx.setBroadcastVariable("name", Arrays.asList(1, 2, 3, 4));
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-core/src/test/java/org/apache/flink/api/common/operators/base/FlatMapOperatorCollectionTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/operators/base/FlatMapOperatorCollectionTest.java b/flink-core/src/test/java/org/apache/flink/api/common/operators/base/FlatMapOperatorCollectionTest.java
index d231455..734324b 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/operators/base/FlatMapOperatorCollectionTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/operators/base/FlatMapOperatorCollectionTest.java
@@ -19,6 +19,7 @@
package org.apache.flink.api.common.operators.base;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
@@ -33,6 +34,7 @@ import org.junit.Test;
import java.io.Serializable;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
@SuppressWarnings("serial")
@@ -72,7 +74,7 @@ public class FlatMapOperatorCollectionTest implements Serializable {
}
// run on collections
final List<String> result = getTestFlatMapOperator(udf)
- .executeOnCollections(input, new RuntimeUDFContext("Test UDF", 4, 0, null, executionConfig), executionConfig);
+ .executeOnCollections(input, new RuntimeUDFContext("Test UDF", 4, 0, null, executionConfig, new HashMap<String, Accumulator<?, ?>>()), executionConfig);
Assert.assertEquals(input.size(), result.size());
Assert.assertEquals(input, result);
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-core/src/test/java/org/apache/flink/api/common/operators/base/JoinOperatorBaseTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/operators/base/JoinOperatorBaseTest.java b/flink-core/src/test/java/org/apache/flink/api/common/operators/base/JoinOperatorBaseTest.java
index 54975b4..98f75bc 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/operators/base/JoinOperatorBaseTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/operators/base/JoinOperatorBaseTest.java
@@ -21,6 +21,7 @@ package org.apache.flink.api.common.operators.base;
import static org.junit.Assert.*;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.functions.FlatJoinFunction;
import org.apache.flink.api.common.functions.RichFlatJoinFunction;
import org.apache.flink.api.common.functions.util.RuntimeUDFContext;
@@ -33,6 +34,7 @@ import org.junit.Test;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.HashMap;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -114,11 +116,12 @@ public class JoinOperatorBaseTest implements Serializable {
try {
+ final HashMap<String, Accumulator<?, ?>> accumulatorMap = new HashMap<String, Accumulator<?, ?>>();
ExecutionConfig executionConfig = new ExecutionConfig();
executionConfig.disableObjectReuse();
- List<Integer> resultSafe = base.executeOnCollections(inputData1, inputData2, new RuntimeUDFContext(taskName, 1, 0, null, executionConfig), executionConfig);
+ List<Integer> resultSafe = base.executeOnCollections(inputData1, inputData2, new RuntimeUDFContext(taskName, 1, 0, null, executionConfig, accumulatorMap), executionConfig);
executionConfig.enableObjectReuse();
- List<Integer> resultRegular = base.executeOnCollections(inputData1, inputData2, new RuntimeUDFContext(taskName, 1, 0, null, executionConfig), executionConfig);
+ List<Integer> resultRegular = base.executeOnCollections(inputData1, inputData2, new RuntimeUDFContext(taskName, 1, 0, null, executionConfig, accumulatorMap), executionConfig);
assertEquals(expected, resultSafe);
assertEquals(expected, resultRegular);
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-core/src/test/java/org/apache/flink/api/common/operators/base/MapOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/operators/base/MapOperatorTest.java b/flink-core/src/test/java/org/apache/flink/api/common/operators/base/MapOperatorTest.java
index 8e07f07..2c98a17 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/operators/base/MapOperatorTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/operators/base/MapOperatorTest.java
@@ -22,10 +22,12 @@ import static org.junit.Assert.*;
import static java.util.Arrays.asList;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
@@ -102,11 +104,12 @@ public class MapOperatorTest implements java.io.Serializable {
parser, new UnaryOperatorInformation<String, Integer>(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO), taskName);
List<String> input = new ArrayList<String>(asList("1", "2", "3", "4", "5", "6"));
+ final HashMap<String, Accumulator<?, ?>> accumulatorMap = new HashMap<String, Accumulator<?, ?>>();
ExecutionConfig executionConfig = new ExecutionConfig();
executionConfig.disableObjectReuse();
- List<Integer> resultMutableSafe = op.executeOnCollections(input, new RuntimeUDFContext(taskName, 1, 0, null, executionConfig), executionConfig);
+ List<Integer> resultMutableSafe = op.executeOnCollections(input, new RuntimeUDFContext(taskName, 1, 0, null, executionConfig, accumulatorMap), executionConfig);
executionConfig.enableObjectReuse();
- List<Integer> resultRegular = op.executeOnCollections(input, new RuntimeUDFContext(taskName, 1, 0, null, executionConfig), executionConfig);
+ List<Integer> resultRegular = op.executeOnCollections(input, new RuntimeUDFContext(taskName, 1, 0, null, executionConfig, accumulatorMap), executionConfig);
assertEquals(asList(1, 2, 3, 4, 5, 6), resultMutableSafe);
assertEquals(asList(1, 2, 3, 4, 5, 6), resultRegular);
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-core/src/test/java/org/apache/flink/api/common/operators/base/PartitionMapOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/operators/base/PartitionMapOperatorTest.java b/flink-core/src/test/java/org/apache/flink/api/common/operators/base/PartitionMapOperatorTest.java
index 61ba359..28c6d821 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/operators/base/PartitionMapOperatorTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/operators/base/PartitionMapOperatorTest.java
@@ -22,10 +22,12 @@ import static org.junit.Assert.*;
import static java.util.Arrays.asList;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.util.Collector;
import org.apache.flink.api.common.functions.MapPartitionFunction;
import org.apache.flink.api.common.functions.RichMapPartitionFunction;
@@ -78,9 +80,9 @@ public class PartitionMapOperatorTest implements java.io.Serializable {
ExecutionConfig executionConfig = new ExecutionConfig();
executionConfig.disableObjectReuse();
- List<Integer> resultMutableSafe = op.executeOnCollections(input, new RuntimeUDFContext(taskName, 1, 0, null, executionConfig), executionConfig);
+ List<Integer> resultMutableSafe = op.executeOnCollections(input, new RuntimeUDFContext(taskName, 1, 0, null, executionConfig, new HashMap<String, Accumulator<?, ?>>()), executionConfig);
executionConfig.enableObjectReuse();
- List<Integer> resultRegular = op.executeOnCollections(input, new RuntimeUDFContext(taskName, 1, 0, null, executionConfig), executionConfig);
+ List<Integer> resultRegular = op.executeOnCollections(input, new RuntimeUDFContext(taskName, 1, 0, null, executionConfig, new HashMap<String, Accumulator<?, ?>>()), executionConfig);
assertEquals(asList(1, 2, 3, 4, 5, 6), resultMutableSafe);
assertEquals(asList(1, 2, 3, 4, 5, 6), resultRegular);
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-examples/flink-java-examples/src/main/java/org/apache/flink/examples/java/wordcount/WordCount.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-java-examples/src/main/java/org/apache/flink/examples/java/wordcount/WordCount.java b/flink-examples/flink-java-examples/src/main/java/org/apache/flink/examples/java/wordcount/WordCount.java
index 82c3ad8..e6b8418 100644
--- a/flink-examples/flink-java-examples/src/main/java/org/apache/flink/examples/java/wordcount/WordCount.java
+++ b/flink-examples/flink-java-examples/src/main/java/org/apache/flink/examples/java/wordcount/WordCount.java
@@ -60,7 +60,7 @@ public class WordCount {
// set up the execution environment
final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
+
// get input data
DataSet<String> text = getTextDataSet(env);
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-java/src/main/java/org/apache/flink/api/java/DataSet.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/DataSet.java b/flink-java/src/main/java/org/apache/flink/api/java/DataSet.java
index d24a350..c628b04 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/DataSet.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/DataSet.java
@@ -408,14 +408,16 @@ public abstract class DataSet<T> {
JobExecutionResult res = getExecutionEnvironment().execute();
ArrayList<byte[]> accResult = res.getAccumulatorResult(id);
- try {
- return SerializedListAccumulator.deserializeList(accResult, serializer);
- }
- catch (ClassNotFoundException e) {
- throw new RuntimeException("Cannot find type class of collected data type.", e);
- }
- catch (IOException e) {
- throw new RuntimeException("Serialization error while deserializing collected data", e);
+ if (accResult != null) {
+ try {
+ return SerializedListAccumulator.deserializeList(accResult, serializer);
+ } catch (ClassNotFoundException e) {
+ throw new RuntimeException("Cannot find type class of collected data type.", e);
+ } catch (IOException e) {
+ throw new RuntimeException("Serialization error while deserializing collected data", e);
+ }
+ } else {
+ throw new RuntimeException("The call to collect() could not retrieve the DataSet.");
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-java/src/main/java/org/apache/flink/api/java/Utils.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/Utils.java b/flink-java/src/main/java/org/apache/flink/api/java/Utils.java
index dd1d6d2..a1e3d25 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/Utils.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/Utils.java
@@ -114,13 +114,18 @@ public class Utils {
@Override
public void open(Configuration parameters) throws Exception {
this.accumulator = new SerializedListAccumulator<T>();
- getRuntimeContext().addAccumulator(id, accumulator);
}
@Override
public void flatMap(T value, Collector<T> out) throws Exception {
accumulator.add(value, serializer);
}
+
+ @Override
+ public void close() throws Exception {
+ // Important: should only be added in close method to minimize traffic of accumulators
+ getRuntimeContext().addAccumulator(id, accumulator);
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-java/src/test/java/org/apache/flink/api/common/operators/base/CoGroupOperatorCollectionTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/CoGroupOperatorCollectionTest.java b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/CoGroupOperatorCollectionTest.java
index 0aa6097..a178a47 100644
--- a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/CoGroupOperatorCollectionTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/CoGroupOperatorCollectionTest.java
@@ -19,6 +19,7 @@
package org.apache.flink.api.common.operators.base;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.functions.RichCoGroupFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
@@ -35,6 +36,7 @@ import org.junit.Test;
import java.io.Serializable;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -68,7 +70,8 @@ public class CoGroupOperatorCollectionTest implements Serializable {
);
ExecutionConfig executionConfig = new ExecutionConfig();
- final RuntimeContext ctx = new RuntimeUDFContext("Test UDF", 4, 0, null, executionConfig);
+ final HashMap<String, Accumulator<?, ?>> accumulators = new HashMap<String, Accumulator<?, ?>>();
+ final RuntimeContext ctx = new RuntimeUDFContext("Test UDF", 4, 0, null, executionConfig, accumulators);
{
SumCoGroup udf1 = new SumCoGroup();
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-java/src/test/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorTest.java b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorTest.java
index 447c8c5..08f4acd 100644
--- a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/GroupReduceOperatorTest.java
@@ -19,6 +19,7 @@
package org.apache.flink.api.common.operators.base;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.functions.GroupReduceFunction;
import org.apache.flink.api.common.functions.RichGroupReduceFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
@@ -31,6 +32,7 @@ import org.apache.flink.util.Collector;
import org.junit.Test;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
@@ -161,9 +163,9 @@ public class GroupReduceOperatorTest implements java.io.Serializable {
ExecutionConfig executionConfig = new ExecutionConfig();
executionConfig.disableObjectReuse();
- List<Tuple2<String, Integer>> resultMutableSafe = op.executeOnCollections(input, new RuntimeUDFContext(taskName, 1, 0, null, executionConfig), executionConfig);
+ List<Tuple2<String, Integer>> resultMutableSafe = op.executeOnCollections(input, new RuntimeUDFContext(taskName, 1, 0, null, executionConfig, new HashMap<String, Accumulator<?, ?>>()), executionConfig);
executionConfig.enableObjectReuse();
- List<Tuple2<String, Integer>> resultRegular = op.executeOnCollections(input, new RuntimeUDFContext(taskName, 1, 0, null, executionConfig), executionConfig);
+ List<Tuple2<String, Integer>> resultRegular = op.executeOnCollections(input, new RuntimeUDFContext(taskName, 1, 0, null, executionConfig, new HashMap<String, Accumulator<?, ?>>()), executionConfig);
Set<Tuple2<String, Integer>> resultSetMutableSafe = new HashSet<Tuple2<String, Integer>>(resultMutableSafe);
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-java/src/test/java/org/apache/flink/api/common/operators/base/JoinOperatorBaseTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/JoinOperatorBaseTest.java b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/JoinOperatorBaseTest.java
index 1b38281..21fcfb3 100644
--- a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/JoinOperatorBaseTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/JoinOperatorBaseTest.java
@@ -21,6 +21,7 @@ package org.apache.flink.api.common.operators.base;
import static org.junit.Assert.*;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.functions.FlatJoinFunction;
import org.apache.flink.api.common.functions.util.RuntimeUDFContext;
import org.apache.flink.api.common.operators.BinaryOperatorInformation;
@@ -33,6 +34,7 @@ import org.junit.Test;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -103,9 +105,9 @@ public class JoinOperatorBaseTest implements Serializable {
try {
ExecutionConfig executionConfig = new ExecutionConfig();
executionConfig.disableObjectReuse();
- List<Tuple2<Double, String>> resultSafe = base.executeOnCollections(inputData1, inputData2, new RuntimeUDFContext("op", 1, 0, null, executionConfig), executionConfig);
+ List<Tuple2<Double, String>> resultSafe = base.executeOnCollections(inputData1, inputData2, new RuntimeUDFContext("op", 1, 0, null, executionConfig, new HashMap<String, Accumulator<?, ?>>()), executionConfig);
executionConfig.enableObjectReuse();
- List<Tuple2<Double, String>> resultRegular = base.executeOnCollections(inputData1, inputData2, new RuntimeUDFContext("op", 1, 0, null, executionConfig), executionConfig);
+ List<Tuple2<Double, String>> resultRegular = base.executeOnCollections(inputData1, inputData2, new RuntimeUDFContext("op", 1, 0, null, executionConfig, new HashMap<String, Accumulator<?, ?>>()), executionConfig);
assertEquals(expected, new HashSet<Tuple2<Double, String>>(resultSafe));
assertEquals(expected, new HashSet<Tuple2<Double, String>>(resultRegular));
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-java/src/test/java/org/apache/flink/api/common/operators/base/ReduceOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/ReduceOperatorTest.java b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/ReduceOperatorTest.java
index 4e1eebd..7cd9771 100644
--- a/flink-java/src/test/java/org/apache/flink/api/common/operators/base/ReduceOperatorTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/common/operators/base/ReduceOperatorTest.java
@@ -19,6 +19,7 @@
package org.apache.flink.api.common.operators.base;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichReduceFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
@@ -30,6 +31,7 @@ import org.apache.flink.configuration.Configuration;
import org.junit.Test;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -138,9 +140,9 @@ public class ReduceOperatorTest implements java.io.Serializable {
ExecutionConfig executionConfig = new ExecutionConfig();
executionConfig.disableObjectReuse();
- List<Tuple2<String, Integer>> resultMutableSafe = op.executeOnCollections(input, new RuntimeUDFContext(taskName, 1, 0, null, executionConfig), executionConfig);
+ List<Tuple2<String, Integer>> resultMutableSafe = op.executeOnCollections(input, new RuntimeUDFContext(taskName, 1, 0, null, executionConfig, new HashMap<String, Accumulator<?, ?>>()), executionConfig);
executionConfig.enableObjectReuse();
- List<Tuple2<String, Integer>> resultRegular = op.executeOnCollections(input, new RuntimeUDFContext(taskName, 1, 0, null, executionConfig), executionConfig);
+ List<Tuple2<String, Integer>> resultRegular = op.executeOnCollections(input, new RuntimeUDFContext(taskName, 1, 0, null, executionConfig, new HashMap<String, Accumulator<?, ?>>()), executionConfig);
Set<Tuple2<String, Integer>> resultSetMutableSafe = new HashSet<Tuple2<String, Integer>>(resultMutableSafe);
Set<Tuple2<String, Integer>> resultSetRegular = new HashSet<Tuple2<String, Integer>>(resultRegular);
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-optimizer/src/main/java/org/apache/flink/optimizer/Optimizer.java
----------------------------------------------------------------------
diff --git a/flink-optimizer/src/main/java/org/apache/flink/optimizer/Optimizer.java b/flink-optimizer/src/main/java/org/apache/flink/optimizer/Optimizer.java
index a2b78ed..6f41c29 100644
--- a/flink-optimizer/src/main/java/org/apache/flink/optimizer/Optimizer.java
+++ b/flink-optimizer/src/main/java/org/apache/flink/optimizer/Optimizer.java
@@ -545,7 +545,7 @@ public class Optimizer {
// ------------------------------------------------------------------------
private OptimizerPostPass getPostPassFromPlan(Plan program) {
- final String className = program.getPostPassClassName();
+ final String className = program.getPostPassClassName();
if (className == null) {
throw new CompilerException("Optimizer Post Pass class description is null");
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorEvent.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorEvent.java b/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorEvent.java
deleted file mode 100644
index ad7fabd..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorEvent.java
+++ /dev/null
@@ -1,49 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.accumulators;
-
-import java.io.IOException;
-import java.util.Map;
-
-import org.apache.flink.api.common.accumulators.Accumulator;
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.runtime.util.SerializedValue;
-
-/**
- * This class encapsulates a map of accumulators for a single job. It is used
- * for the transfer from TaskManagers to the JobManager and from the JobManager
- * to the Client.
- */
-public class AccumulatorEvent extends SerializedValue<Map<String, Accumulator<?, ?>>> {
-
- private static final long serialVersionUID = 8965894516006882735L;
-
- /** JobID for the target job */
- private final JobID jobID;
-
-
- public AccumulatorEvent(JobID jobID, Map<String, Accumulator<?, ?>> accumulators) throws IOException {
- super(accumulators);
- this.jobID = jobID;
- }
-
- public JobID getJobID() {
- return this.jobID;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorRegistry.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorRegistry.java b/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorRegistry.java
new file mode 100644
index 0000000..0ef3650
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorRegistry.java
@@ -0,0 +1,147 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.accumulators;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.api.common.accumulators.LongCounter;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+
+/**
+ * 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;
+ protected final ExecutionAttemptID taskID;
+
+ /* Flink's internal Accumulator values stored for the executing task. */
+ private final Map<Metric, Accumulator<?, ?>> flinkAccumulators =
+ new HashMap<Metric, Accumulator<?, ?>>();
+
+ /* User-defined Accumulator values stored for the executing task. */
+ private final Map<String, Accumulator<?, ?>> userAccumulators =
+ Collections.synchronizedMap(new HashMap<String, Accumulator<?, ?>>());
+
+ /* The reporter reference that is handed to the reporting tasks. */
+ private final ReadWriteReporter reporter;
+
+ /**
+ * Flink metrics supported
+ */
+ public enum Metric {
+ NUM_RECORDS_IN,
+ NUM_RECORDS_OUT,
+ NUM_BYTES_IN,
+ NUM_BYTES_OUT
+ }
+
+
+ public AccumulatorRegistry(JobID jobID, ExecutionAttemptID taskID) {
+ this.jobID = jobID;
+ this.taskID = taskID;
+ this.reporter = new ReadWriteReporter(flinkAccumulators);
+ }
+
+ /**
+ * 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;
+ }
+ }
+
+ /**
+ * Gets the map for user-defined accumulators.
+ */
+ public Map<String, Accumulator<?, ?>> getUserMap() {
+ return userAccumulators;
+ }
+
+ /**
+ * Gets the reporter for flink internal metrics.
+ */
+ public Reporter getReadWriteReporter() {
+ return reporter;
+ }
+
+ /**
+ * Interface for Flink's internal accumulators.
+ */
+ public interface Reporter {
+ void reportNumRecordsIn(long value);
+ void reportNumRecordsOut(long value);
+ void reportNumBytesIn(long value);
+ void reportNumBytesOut(long value);
+ }
+
+ /**
+ * 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);
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorSnapshot.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorSnapshot.java b/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorSnapshot.java
new file mode 100644
index 0000000..a6288f0
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorSnapshot.java
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.accumulators;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+import org.apache.flink.runtime.util.SerializedValue;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Map;
+
+/**
+ * This class encapsulates a map of accumulators for a single task. It is used
+ * for the transfer from TaskManagers to the JobManager and from the JobManager
+ * to the Client.
+ */
+public class AccumulatorSnapshot implements Serializable {
+
+ private static final long serialVersionUID = 42L;
+
+ private final JobID jobID;
+ private final ExecutionAttemptID executionAttemptID;
+
+ /**
+ * Flink internal accumulators which can be serialized using the system class loader.
+ */
+ private final Map<AccumulatorRegistry.Metric, Accumulator<?, ?>> flinkAccumulators;
+
+ /**
+ * Serialized user accumulators which may require the custom user class loader.
+ */
+ private final SerializedValue<Map<String, Accumulator<?, ?>>> userAccumulators;
+
+ 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 = flinkAccumulators;
+ this.userAccumulators = new SerializedValue<Map<String, Accumulator<?, ?>>>(userAccumulators);
+ }
+
+ public JobID getJobID() {
+ return jobID;
+ }
+
+ public ExecutionAttemptID getExecutionAttemptID() {
+ return executionAttemptID;
+ }
+
+ /**
+ * Gets the Flink (internal) accumulators values.
+ * @return the serialized map
+ */
+ public Map<AccumulatorRegistry.Metric, Accumulator<?, ?>> getFlinkAccumulators() {
+ return flinkAccumulators;
+ }
+
+ /**
+ * Gets the user-defined accumulators values.
+ * @return the serialized map
+ */
+ public Map<String, Accumulator<?, ?>> deserializeUserAccumulators(ClassLoader classLoader) throws IOException, ClassNotFoundException {
+ return userAccumulators.deserializeValue(classLoader);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/execution/Environment.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/Environment.java b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/Environment.java
index 755f1ad..c561869 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/Environment.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/Environment.java
@@ -18,9 +18,9 @@
package org.apache.flink.runtime.execution;
-import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.broadcast.BroadcastVariableManager;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.io.disk.iomanager.IOManager;
@@ -142,11 +142,10 @@ public interface Environment {
BroadcastVariableManager getBroadcastVariableManager();
/**
- * Reports the given set of accumulators to the JobManager.
- *
- * @param accumulators The accumulators to report.
+ * Return the registry for accumulators which are periodically sent to the job manager.
+ * @return the registry
*/
- void reportAccumulators(Map<String, Accumulator<?, ?>> accumulators);
+ AccumulatorRegistry getAccumulatorRegistry();
/**
* Confirms that the invokable has successfully completed all steps it needed to
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManager.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManager.java
index 848f619..88be5e1 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/execution/librarycache/BlobLibraryCacheManager.java
@@ -46,7 +46,7 @@ import com.google.common.base.Preconditions;
* For each job graph that is submitted to the system the library cache manager maintains
* a set of libraries (typically JAR files) which the job requires to run. The library cache manager
* caches library files in order to avoid unnecessary retransmission of data. It is based on a singleton
- * programming pattern, so there exists at most on library manager at a time.
+ * programming pattern, so there exists at most one library manager at a time.
*/
public final class BlobLibraryCacheManager extends TimerTask implements LibraryCacheManager {
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
index af67c3f..3b836ad 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
@@ -20,7 +20,9 @@ package org.apache.flink.runtime.executiongraph;
import akka.dispatch.OnComplete;
import akka.dispatch.OnFailure;
+import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.runtime.JobException;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.deployment.InputChannelDeploymentDescriptor;
import org.apache.flink.runtime.deployment.PartialInputChannelDeploymentDescriptor;
import org.apache.flink.runtime.deployment.ResultPartitionLocation;
@@ -54,6 +56,7 @@ import scala.concurrent.duration.FiniteDuration;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeoutException;
@@ -133,6 +136,15 @@ public class Execution implements Serializable {
@SuppressWarnings("NonSerializableFieldInSerializableClass")
private ExecutionContext executionContext;
+ /* Lock for updating the accumulators atomically. */
+ private final Object accumulatorLock = new Object();
+
+ /* Continuously updated map of user-defined accumulators */
+ private volatile Map<String, Accumulator<?, ?>> userAccumulators;
+
+ /* Continuously updated map of internal accumulators */
+ private volatile Map<AccumulatorRegistry.Metric, Accumulator<?, ?>> flinkAccumulators;
+
// --------------------------------------------------------------------------------------------
public Execution(
@@ -593,6 +605,10 @@ public class Execution implements Serializable {
}
void markFinished() {
+ markFinished(null, null);
+ }
+
+ void markFinished(Map<AccumulatorRegistry.Metric, Accumulator<?, ?>> flinkAccumulators, Map<String, Accumulator<?, ?>> userAccumulators) {
// this call usually comes during RUNNING, but may also come while still in deploying (very fast tasks!)
while (true) {
@@ -613,6 +629,11 @@ public class Execution implements Serializable {
}
}
+ synchronized (accumulatorLock) {
+ this.flinkAccumulators = flinkAccumulators;
+ this.userAccumulators = userAccumulators;
+ }
+
assignedResource.releaseSlot();
vertex.getExecutionGraph().deregisterExecution(this);
}
@@ -935,6 +956,28 @@ public class Execution implements Serializable {
return vertex.getSimpleName() + " - execution #" + attemptNumber;
}
+ /**
+ * 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;
+ }
+ }
+ }
+ public Map<String, Accumulator<?, ?>> getUserAccumulators() {
+ return userAccumulators;
+ }
+
+ public Map<AccumulatorRegistry.Metric, Accumulator<?, ?>> getFlinkAccumulators() {
+ return flinkAccumulators;
+ }
+
@Override
public String toString() {
return String.format("Attempt #%d (%s) @ %s - [%s]", attemptNumber, vertex.getSimpleName(),
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
index 47b7ae2..6d2262c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
@@ -22,8 +22,12 @@ import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.api.common.accumulators.AccumulatorHelper;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.JobException;
+import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.blob.BlobKey;
import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
import org.apache.flink.runtime.execution.ExecutionState;
@@ -38,6 +42,7 @@ import org.apache.flink.runtime.jobmanager.scheduler.Scheduler;
import org.apache.flink.runtime.messages.ExecutionGraphMessages;
import org.apache.flink.runtime.taskmanager.TaskExecutionState;
import org.apache.flink.runtime.util.SerializableObject;
+import org.apache.flink.runtime.util.SerializedValue;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.InstantiationUtil;
@@ -47,10 +52,12 @@ import org.slf4j.LoggerFactory;
import scala.concurrent.ExecutionContext;
import scala.concurrent.duration.FiniteDuration;
+import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -128,6 +135,29 @@ public class ExecutionGraph implements Serializable {
/** The currently executed tasks, for callbacks */
private final ConcurrentHashMap<ExecutionAttemptID, Execution> currentExecutions;
+ /**
+ * 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 = accumulatorSnapshot.getFlinkAccumulators();
+ Map<String, Accumulator<?, ?>> userAccumulators;
+ try {
+ userAccumulators = accumulatorSnapshot.deserializeUserAccumulators(userClassLoader);
+
+ ExecutionAttemptID execID = accumulatorSnapshot.getExecutionAttemptID();
+ Execution execution = currentExecutions.get(execID);
+ if (execution != null) {
+ execution.setAccumulators(flinkAccumulators, userAccumulators);
+ } else {
+ LOG.warn("Received accumulator result for unknown execution {}.", execID);
+ }
+ } catch (Exception e) {
+ LOG.error("Cannot update accumulators for job " + jobID, e);
+ }
+ }
+
/** A list of all libraries required during the job execution. Libraries have to be stored
* inside the BlobService and are referenced via the BLOB keys. */
private final List<BlobKey> requiredJarFiles;
@@ -485,6 +515,57 @@ public class ExecutionGraph implements Serializable {
return executionContext;
}
+ /**
+ * Gets the internal flink accumulator map of maps which contains some metrics.
+ * @return A map of accumulators for every executed task.
+ */
+ public Map<ExecutionAttemptID, Map<AccumulatorRegistry.Metric, Accumulator<?,?>>> getFlinkAccumulators() {
+ Map<ExecutionAttemptID, Map<AccumulatorRegistry.Metric, Accumulator<?, ?>>> flinkAccumulators =
+ new HashMap<ExecutionAttemptID, Map<AccumulatorRegistry.Metric, Accumulator<?, ?>>>();
+
+ for (ExecutionVertex vertex : getAllExecutionVertices()) {
+ Map<AccumulatorRegistry.Metric, Accumulator<?, ?>> taskAccs = vertex.getCurrentExecutionAttempt().getFlinkAccumulators();
+ flinkAccumulators.put(vertex.getCurrentExecutionAttempt().getAttemptId(), taskAccs);
+ }
+
+ return flinkAccumulators;
+ }
+
+ /**
+ * 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;
+ }
+
+ /**
+ * 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;
+ }
+
// --------------------------------------------------------------------------------------------
// Actions
// --------------------------------------------------------------------------------------------
@@ -791,14 +872,31 @@ public class ExecutionGraph implements Serializable {
// Callbacks and Callback Utilities
// --------------------------------------------------------------------------------------------
+ /**
+ * Updates the state of the Task and sets the final accumulator results.
+ * @param state
+ * @return
+ */
public boolean updateState(TaskExecutionState state) {
Execution attempt = this.currentExecutions.get(state.getID());
if (attempt != null) {
+
switch (state.getExecutionState()) {
case RUNNING:
return attempt.switchToRunning();
case FINISHED:
- attempt.markFinished();
+ Map<AccumulatorRegistry.Metric, Accumulator<?, ?>> flinkAccumulators = null;
+ Map<String, Accumulator<?, ?>> userAccumulators = null;
+ try {
+ AccumulatorSnapshot accumulators = state.getAccumulators();
+ flinkAccumulators = accumulators.getFlinkAccumulators();
+ userAccumulators = accumulators.deserializeUserAccumulators(userClassLoader);
+ } catch (Exception e) {
+ // Exceptions would be thrown in the future here
+ LOG.error("Failed to deserialize final accumulator results.", e);
+ }
+
+ attempt.markFinished(flinkAccumulators, userAccumulators);
return true;
case CANCELED:
attempt.cancelingComplete();
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/AbstractReader.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/AbstractReader.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/AbstractReader.java
index 96b6f99..90564a8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/AbstractReader.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/AbstractReader.java
@@ -147,4 +147,5 @@ public abstract class AbstractReader implements ReaderBase {
return ++currentNumberOfEndOfSuperstepEvents == inputGate.getNumberOfInputChannels();
}
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/AbstractRecordReader.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/AbstractRecordReader.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/AbstractRecordReader.java
index 56e5d33..7aa57a8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/AbstractRecordReader.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/AbstractRecordReader.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.io.network.api.reader;
import org.apache.flink.core.io.IOReadableWritable;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.io.network.api.serialization.RecordDeserializer;
import org.apache.flink.runtime.io.network.api.serialization.RecordDeserializer.DeserializationResult;
import org.apache.flink.runtime.io.network.api.serialization.SpillingAdaptiveSpanningRecordDeserializer;
@@ -64,7 +65,9 @@ abstract class AbstractRecordReader<T extends IOReadableWritable> extends Abstra
DeserializationResult result = currentRecordDeserializer.getNextRecord(target);
if (result.isBufferConsumed()) {
- currentRecordDeserializer.getCurrentBuffer().recycle();
+ final Buffer currentBuffer = currentRecordDeserializer.getCurrentBuffer();
+
+ currentBuffer.recycle();
currentRecordDeserializer = null;
}
@@ -89,7 +92,7 @@ abstract class AbstractRecordReader<T extends IOReadableWritable> extends Abstra
+ "If you are using custom serialization code (Writable or Value types), check their "
+ "serialization routines. In the case of Kryo, check the respective Kryo serializer.");
}
-
+
if (handleEvent(bufferOrEvent.getEvent())) {
if (inputGate.isFinished()) {
isFinished = true;
@@ -112,4 +115,11 @@ abstract class AbstractRecordReader<T extends IOReadableWritable> extends Abstra
}
}
}
+
+ @Override
+ public void setReporter(AccumulatorRegistry.Reporter reporter) {
+ for (RecordDeserializer<?> deserializer : recordDeserializers) {
+ deserializer.setReporter(reporter);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/BufferReader.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/BufferReader.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/BufferReader.java
index ca59609..debb352 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/BufferReader.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/BufferReader.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.io.network.api.reader;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.io.network.partition.consumer.BufferOrEvent;
import org.apache.flink.runtime.io.network.partition.consumer.InputGate;
@@ -47,4 +48,9 @@ public final class BufferReader extends AbstractReader {
}
}
}
+
+ @Override
+ public void setReporter(AccumulatorRegistry.Reporter reporter) {
+
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/ReaderBase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/ReaderBase.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/ReaderBase.java
index 2a0a6df..9f8ae20 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/ReaderBase.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/ReaderBase.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.io.network.api.reader;
import java.io.IOException;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.event.task.TaskEvent;
import org.apache.flink.runtime.util.event.EventListener;
@@ -51,4 +52,9 @@ public interface ReaderBase {
boolean hasReachedEndOfSuperstep();
+ /**
+ * Setter for the reporter, e.g. for the number of records emitted and the number of bytes read.
+ */
+ void setReporter(AccumulatorRegistry.Reporter reporter);
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/RecordReader.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/RecordReader.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/RecordReader.java
index b1395e3..d45920e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/RecordReader.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/reader/RecordReader.java
@@ -80,4 +80,5 @@ public class RecordReader<T extends IOReadableWritable> extends AbstractRecordRe
throw new RuntimeException("Cannot instantiate class " + recordType.getName(), e);
}
}
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/AdaptiveSpanningRecordDeserializer.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/AdaptiveSpanningRecordDeserializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/AdaptiveSpanningRecordDeserializer.java
index 28bcf4a..ec9f4fd 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/AdaptiveSpanningRecordDeserializer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/AdaptiveSpanningRecordDeserializer.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.io.network.api.serialization;
import org.apache.flink.core.io.IOReadableWritable;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.util.DataInputDeserializer;
import org.apache.flink.runtime.util.DataOutputSerializer;
@@ -45,6 +46,8 @@ public class AdaptiveSpanningRecordDeserializer<T extends IOReadableWritable> im
private Buffer currentBuffer;
+ private AccumulatorRegistry.Reporter reporter;
+
public AdaptiveSpanningRecordDeserializer() {
this.nonSpanningWrapper = new NonSpanningWrapper();
this.spanningWrapper = new SpanningWrapper();
@@ -90,10 +93,18 @@ public class AdaptiveSpanningRecordDeserializer<T extends IOReadableWritable> im
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);
+ }
+
return (this.nonSpanningWrapper.remaining() == 0) ?
DeserializationResult.LAST_RECORD_FROM_BUFFER :
DeserializationResult.INTERMEDIATE_RECORD_FROM_BUFFER;
@@ -117,6 +128,10 @@ public class AdaptiveSpanningRecordDeserializer<T extends IOReadableWritable> im
// get the full record
target.read(this.spanningWrapper);
+ if (reporter != null) {
+ reporter.reportNumRecordsIn(1);
+ }
+
// move the remainder to the non-spanning wrapper
// this does not copy it, only sets the memory segment
this.spanningWrapper.moveRemainderToNonSpanningDeserializer(this.nonSpanningWrapper);
@@ -144,6 +159,12 @@ public class AdaptiveSpanningRecordDeserializer<T extends IOReadableWritable> im
return this.nonSpanningWrapper.remaining() > 0 || this.spanningWrapper.getNumGatheredBytes() > 0;
}
+ @Override
+ public void setReporter(AccumulatorRegistry.Reporter reporter) {
+ this.reporter = reporter;
+ this.spanningWrapper.setReporter(reporter);
+ }
+
// -----------------------------------------------------------------------------------------------------------------
private static final class NonSpanningWrapper implements DataInputView {
@@ -426,6 +447,8 @@ public class AdaptiveSpanningRecordDeserializer<T extends IOReadableWritable> im
private int recordLimit;
+ private AccumulatorRegistry.Reporter reporter;
+
public SpanningWrapper() {
this.lengthBuffer = ByteBuffer.allocate(4);
this.lengthBuffer.order(ByteOrder.BIG_ENDIAN);
@@ -463,6 +486,10 @@ public class AdaptiveSpanningRecordDeserializer<T extends IOReadableWritable> im
} else {
this.recordLength = this.lengthBuffer.getInt(0);
+ if (reporter != null) {
+ reporter.reportNumBytesIn(this.recordLength);
+ }
+
this.lengthBuffer.clear();
segmentPosition = toPut;
}
@@ -607,5 +634,9 @@ public class AdaptiveSpanningRecordDeserializer<T extends IOReadableWritable> im
public int read(byte[] b) throws IOException {
return this.serializationReadBuffer.read(b);
}
+
+ public void setReporter(AccumulatorRegistry.Reporter reporter) {
+ this.reporter = reporter;
+ }
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/RecordDeserializer.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/RecordDeserializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/RecordDeserializer.java
index dd8ea06..e4c7890 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/RecordDeserializer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/RecordDeserializer.java
@@ -23,6 +23,7 @@ import java.io.IOException;
import org.apache.flink.core.io.IOReadableWritable;
import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.io.network.buffer.Buffer;
/**
@@ -64,4 +65,9 @@ public interface RecordDeserializer<T extends IOReadableWritable> {
void clear();
boolean hasUnfinishedData();
+
+ /**
+ * Setter for the reporter, e.g. for the number of records emitted and the number of bytes read.
+ */
+ void setReporter(AccumulatorRegistry.Reporter reporter);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/RecordSerializer.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/RecordSerializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/RecordSerializer.java
index bb6db55..e9f339a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/RecordSerializer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/RecordSerializer.java
@@ -22,6 +22,7 @@ package org.apache.flink.runtime.io.network.api.serialization;
import java.io.IOException;
import org.apache.flink.core.io.IOReadableWritable;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.io.network.buffer.Buffer;
/**
@@ -63,4 +64,9 @@ public interface RecordSerializer<T extends IOReadableWritable> {
void clear();
boolean hasData();
+
+ /**
+ * Setter for the reporter, e.g. for the number of records emitted and the number of bytes read.
+ */
+ void setReporter(AccumulatorRegistry.Reporter reporter);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpanningRecordSerializer.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpanningRecordSerializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpanningRecordSerializer.java
index 38e130d..f163e05 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpanningRecordSerializer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpanningRecordSerializer.java
@@ -24,6 +24,7 @@ import java.nio.ByteOrder;
import org.apache.flink.core.io.IOReadableWritable;
import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.util.DataOutputSerializer;
@@ -50,6 +51,8 @@ public class SpanningRecordSerializer<T extends IOReadableWritable> implements R
/** Limit of current {@link MemorySegment} of target buffer */
private int limit;
+ private AccumulatorRegistry.Reporter reporter;
+
public SpanningRecordSerializer() {
this.serializationBuffer = new DataOutputSerializer(128);
@@ -75,7 +78,13 @@ public class SpanningRecordSerializer<T extends IOReadableWritable> implements R
// write data and length
record.write(this.serializationBuffer);
- this.lengthBuffer.putInt(0, this.serializationBuffer.length());
+ int len = this.serializationBuffer.length();
+ this.lengthBuffer.putInt(0, len);
+
+ if (reporter != null) {
+ reporter.reportNumBytesOut(len);
+ reporter.reportNumRecordsOut(1);
+ }
this.dataBuffer = this.serializationBuffer.wrapAsByteBuffer();
@@ -173,4 +182,9 @@ public class SpanningRecordSerializer<T extends IOReadableWritable> implements R
// either data in current target buffer or intermediate buffers
return this.position > 0 || (this.lengthBuffer.hasRemaining() || this.dataBuffer.hasRemaining());
}
+
+ @Override
+ public void setReporter(AccumulatorRegistry.Reporter reporter) {
+ this.reporter = reporter;
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpillingAdaptiveSpanningRecordDeserializer.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpillingAdaptiveSpanningRecordDeserializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpillingAdaptiveSpanningRecordDeserializer.java
index 6b0d836..f3e4892 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpillingAdaptiveSpanningRecordDeserializer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/serialization/SpillingAdaptiveSpanningRecordDeserializer.java
@@ -24,6 +24,7 @@ import org.apache.flink.core.io.IOReadableWritable;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.InputViewDataInputStreamWrapper;
import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.util.DataInputDeserializer;
import org.apache.flink.util.StringUtils;
@@ -61,6 +62,8 @@ public class SpillingAdaptiveSpanningRecordDeserializer<T extends IOReadableWrit
private Buffer currentBuffer;
+ private AccumulatorRegistry.Reporter reporter;
+
public SpillingAdaptiveSpanningRecordDeserializer() {
String tempDirString = GlobalConfiguration.getString(
@@ -112,11 +115,19 @@ public class SpillingAdaptiveSpanningRecordDeserializer<T extends IOReadableWrit
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
try {
target.read(this.nonSpanningWrapper);
+ if (reporter != null) {
+ reporter.reportNumRecordsIn(1);
+ }
+
int remaining = this.nonSpanningWrapper.remaining();
if (remaining > 0) {
return DeserializationResult.INTERMEDIATE_RECORD_FROM_BUFFER;
@@ -151,6 +162,10 @@ public class SpillingAdaptiveSpanningRecordDeserializer<T extends IOReadableWrit
if (this.spanningWrapper.hasFullRecord()) {
// get the full record
target.read(this.spanningWrapper.getInputView());
+
+ if (reporter != null) {
+ reporter.reportNumRecordsIn(1);
+ }
// move the remainder to the non-spanning wrapper
// this does not copy it, only sets the memory segment
@@ -176,6 +191,12 @@ public class SpillingAdaptiveSpanningRecordDeserializer<T extends IOReadableWrit
return this.nonSpanningWrapper.remaining() > 0 || this.spanningWrapper.getNumGatheredBytes() > 0;
}
+ @Override
+ public void setReporter(AccumulatorRegistry.Reporter reporter) {
+ this.reporter = reporter;
+ this.spanningWrapper.setReporter(reporter);
+ }
+
// -----------------------------------------------------------------------------------------------------------------
private static final class NonSpanningWrapper implements DataInputView {
@@ -469,7 +490,9 @@ public class SpillingAdaptiveSpanningRecordDeserializer<T extends IOReadableWrit
private File spillFile;
private InputViewDataInputStreamWrapper spillFileReader;
-
+
+ private AccumulatorRegistry.Reporter reporter;
+
public SpanningWrapper(String[] tempDirs) {
this.tempDirs = tempDirs;
@@ -522,6 +545,11 @@ public class SpillingAdaptiveSpanningRecordDeserializer<T extends IOReadableWrit
return;
} else {
this.recordLength = this.lengthBuffer.getInt(0);
+
+ if (reporter != null) {
+ reporter.reportNumBytesIn(recordLength);
+ }
+
this.lengthBuffer.clear();
segmentPosition = toPut;
@@ -652,5 +680,9 @@ public class SpillingAdaptiveSpanningRecordDeserializer<T extends IOReadableWrit
random.nextBytes(bytes);
return StringUtils.byteToHexString(bytes);
}
+
+ public void setReporter(AccumulatorRegistry.Reporter reporter) {
+ this.reporter = reporter;
+ }
}
}
[2/4] flink git commit: [FLINK-2292][FLINK-1573] add live per-task
accumulators
Posted by mx...@apache.org.
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/io/CoRecordReader.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/io/CoRecordReader.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/io/CoRecordReader.java
index cfd27f7..a7139b6 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/io/CoRecordReader.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/io/CoRecordReader.java
@@ -22,6 +22,7 @@ import java.util.LinkedList;
import java.util.concurrent.LinkedBlockingDeque;
import org.apache.flink.core.io.IOReadableWritable;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.io.network.api.reader.AbstractReader;
import org.apache.flink.runtime.io.network.api.reader.MutableRecordReader;
import org.apache.flink.runtime.io.network.api.serialization.AdaptiveSpanningRecordDeserializer;
@@ -256,6 +257,16 @@ public class CoRecordReader<T1 extends IOReadableWritable, T2 extends IOReadable
}
}
+ @Override
+ public void setReporter(AccumulatorRegistry.Reporter reporter) {
+ for (AdaptiveSpanningRecordDeserializer serializer : reader1RecordDeserializers) {
+ serializer.setReporter(reporter);
+ }
+ for (AdaptiveSpanningRecordDeserializer serializer : reader2RecordDeserializers) {
+ serializer.setReporter(reporter);
+ }
+ }
+
private class CoBarrierBuffer extends BarrierBuffer {
private CoBarrierBuffer otherBuffer;
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/io/StreamingAbstractRecordReader.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/io/StreamingAbstractRecordReader.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/io/StreamingAbstractRecordReader.java
index 491dc06..44f9a86 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/io/StreamingAbstractRecordReader.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/io/StreamingAbstractRecordReader.java
@@ -21,6 +21,7 @@ package org.apache.flink.streaming.runtime.io;
import java.io.IOException;
import org.apache.flink.core.io.IOReadableWritable;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.event.task.AbstractEvent;
import org.apache.flink.runtime.io.network.api.reader.AbstractReader;
import org.apache.flink.runtime.io.network.api.reader.ReaderBase;
@@ -57,6 +58,7 @@ public abstract class StreamingAbstractRecordReader<T extends IOReadableWritable
private final BarrierBuffer barrierBuffer;
+
@SuppressWarnings("unchecked")
protected StreamingAbstractRecordReader(InputGate inputGate) {
super(inputGate);
@@ -80,7 +82,8 @@ public abstract class StreamingAbstractRecordReader<T extends IOReadableWritable
DeserializationResult result = currentRecordDeserializer.getNextRecord(target);
if (result.isBufferConsumed()) {
- currentRecordDeserializer.getCurrentBuffer().recycle();
+ Buffer currentBuffer = currentRecordDeserializer.getCurrentBuffer();
+ currentBuffer.recycle();
currentRecordDeserializer = null;
}
@@ -130,4 +133,12 @@ public abstract class StreamingAbstractRecordReader<T extends IOReadableWritable
public void cleanup() throws IOException {
barrierBuffer.cleanup();
}
+
+ @Override
+ public void setReporter(AccumulatorRegistry.Reporter reporter) {
+ for (RecordDeserializer<?> deserializer : recordDeserializers) {
+ deserializer.setReporter(reporter);
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/io/StreamingMutableRecordReader.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/io/StreamingMutableRecordReader.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/io/StreamingMutableRecordReader.java
index ad74004..1356af5 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/io/StreamingMutableRecordReader.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/io/StreamingMutableRecordReader.java
@@ -40,4 +40,5 @@ public class StreamingMutableRecordReader<T extends IOReadableWritable> extends
public void clearBuffers() {
super.clearBuffers();
}
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTask.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTask.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTask.java
index 80239fd..a9ebf5b 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTask.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTask.java
@@ -20,6 +20,7 @@ package org.apache.flink.streaming.runtime.tasks;
import java.io.IOException;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.io.network.partition.consumer.InputGate;
import org.apache.flink.runtime.plugable.DeserializationDelegate;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
@@ -52,6 +53,11 @@ public class OneInputStreamTask<IN, OUT> extends StreamTask<OUT, OneInputStreamO
InputGate inputGate = InputGateFactory.createInputGate(getEnvironment().getAllInputGates());
inputs = new IndexedMutableReader<DeserializationDelegate<StreamRecord<IN>>>(inputGate);
+ AccumulatorRegistry registry = getEnvironment().getAccumulatorRegistry();
+ AccumulatorRegistry.Reporter reporter = registry.getReadWriteReporter();
+
+ inputs.setReporter(reporter);
+
inputs.registerTaskEventListener(getSuperstepListener(), StreamingSuperstep.class);
recordIterator = new IndexedReaderIterator<StreamRecord<IN>>(inputs, inSerializer);
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/OutputHandler.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/OutputHandler.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/OutputHandler.java
index 73f0a89..41ee388 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/OutputHandler.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/OutputHandler.java
@@ -24,7 +24,10 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import com.google.common.base.Preconditions;
+import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.io.network.api.writer.RecordWriter;
import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
import org.apache.flink.runtime.plugable.SerializationDelegate;
@@ -59,7 +62,14 @@ public class OutputHandler<OUT> {
private Map<Integer, StreamConfig> chainedConfigs;
private List<StreamEdge> outEdgesInOrder;
- public OutputHandler(StreamTask<OUT, ?> vertex) {
+ /**
+ * Counters for the number of records emitted and bytes written.
+ */
+ protected AccumulatorRegistry.Reporter reporter;
+
+
+ public OutputHandler(StreamTask<OUT, ?> vertex, Map<String, Accumulator<?,?>> accumulatorMap,
+ AccumulatorRegistry.Reporter reporter) {
// Initialize some fields
this.vertex = vertex;
@@ -75,6 +85,8 @@ public class OutputHandler<OUT> {
this.outEdgesInOrder = configuration.getOutEdgesInOrder(cl);
+ this.reporter = reporter;
+
// We iterate through all the out edges from this job vertex and create
// a stream output
for (StreamEdge outEdge : outEdgesInOrder) {
@@ -82,13 +94,14 @@ public class OutputHandler<OUT> {
outEdge,
outEdge.getTargetId(),
chainedConfigs.get(outEdge.getSourceId()),
- outEdgesInOrder.indexOf(outEdge));
+ outEdgesInOrder.indexOf(outEdge),
+ reporter);
outputMap.put(outEdge, streamOutput);
}
// We create the outer output that will be passed to the first task
// in the chain
- this.outerOutput = createChainedCollector(configuration);
+ this.outerOutput = createChainedCollector(configuration, accumulatorMap);
// Add the head operator to the end of the list
this.chainedOperators.add(vertex.streamOperator);
@@ -121,7 +134,8 @@ public class OutputHandler<OUT> {
* config
*/
@SuppressWarnings({"unchecked", "rawtypes"})
- private <X> Output<X> createChainedCollector(StreamConfig chainedTaskConfig) {
+ private <X> Output<X> createChainedCollector(StreamConfig chainedTaskConfig, Map<String, Accumulator<?,?>> accumulatorMap) {
+ Preconditions.checkNotNull(accumulatorMap);
// We create a wrapper that will encapsulate the chained operators and
@@ -141,7 +155,7 @@ public class OutputHandler<OUT> {
for (StreamEdge outputEdge : chainedTaskConfig.getChainedOutputs(cl)) {
Integer output = outputEdge.getTargetId();
- Collector<?> outCollector = createChainedCollector(chainedConfigs.get(output));
+ Collector<?> outCollector = createChainedCollector(chainedConfigs.get(output), accumulatorMap);
wrapper.addCollector(outCollector, outputEdge);
}
@@ -155,8 +169,8 @@ public class OutputHandler<OUT> {
// operator which will be returned and set it up using the wrapper
OneInputStreamOperator chainableOperator =
chainedTaskConfig.getStreamOperator(vertex.getUserCodeClassLoader());
-
- StreamingRuntimeContext chainedContext = vertex.createRuntimeContext(chainedTaskConfig);
+
+ StreamingRuntimeContext chainedContext = vertex.createRuntimeContext(chainedTaskConfig, accumulatorMap);
vertex.contexts.add(chainedContext);
chainableOperator.setup(wrapper, chainedContext);
@@ -188,7 +202,7 @@ public class OutputHandler<OUT> {
* @return The created StreamOutput
*/
private <T> StreamOutput<T> createStreamOutput(StreamEdge edge, Integer outputVertex,
- StreamConfig upStreamConfig, int outputIndex) {
+ StreamConfig upStreamConfig, int outputIndex, AccumulatorRegistry.Reporter reporter) {
StreamRecordSerializer<T> outSerializer = upStreamConfig
.getTypeSerializerOut1(vertex.userClassLoader);
@@ -207,6 +221,8 @@ public class OutputHandler<OUT> {
RecordWriter<SerializationDelegate<StreamRecord<T>>> output =
RecordWriterFactory.createRecordWriter(bufferWriter, outputPartitioner, upStreamConfig.getBufferTimeout());
+ output.setReporter(reporter);
+
StreamOutput<T> streamOutput = new StreamOutput<T>(output, outSerializationDelegate);
if (LOG.isTraceEnabled()) {
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamIterationHead.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamIterationHead.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamIterationHead.java
index 25fe83d..e5d58d3 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamIterationHead.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamIterationHead.java
@@ -18,10 +18,13 @@
package org.apache.flink.streaming.runtime.tasks;
import java.util.Collection;
+import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.streaming.api.collector.StreamOutput;
import org.apache.flink.streaming.runtime.io.BlockingQueueBroker;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
@@ -46,7 +49,11 @@ public class StreamIterationHead<OUT> extends OneInputStreamTask<OUT, OUT> {
@Override
public void registerInputOutput() {
super.registerInputOutput();
- outputHandler = new OutputHandler<OUT>(this);
+
+ final AccumulatorRegistry registry = getEnvironment().getAccumulatorRegistry();
+ Map<String, Accumulator<?, ?>> accumulatorMap = registry.getUserMap();
+
+ outputHandler = new OutputHandler<OUT>(this, accumulatorMap, outputHandler.reporter);
String iterationId = configuration.getIterationId();
iterationWaitTime = configuration.getIterationWaitTime();
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
index f98ed2d..4ffc8f5 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
@@ -25,10 +25,12 @@ import java.util.Map;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.functors.NotNullPredicate;
+import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.event.task.TaskEvent;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
@@ -87,13 +89,19 @@ public abstract class StreamTask<OUT, O extends StreamOperator<OUT>> extends Abs
streamOperator = configuration.getStreamOperator(userClassLoader);
- outputHandler = new OutputHandler<OUT>(this);
+ // Create and register Accumulators
+ Environment env = getEnvironment();
+ AccumulatorRegistry accumulatorRegistry = env.getAccumulatorRegistry();
+ Map<String, Accumulator<?, ?>> accumulatorMap = accumulatorRegistry.getUserMap();
+ AccumulatorRegistry.Reporter reporter = accumulatorRegistry.getReadWriteReporter();
+
+ outputHandler = new OutputHandler<OUT>(this, accumulatorMap, reporter);
if (streamOperator != null) {
// IterationHead and IterationTail don't have an Operator...
//Create context of the head operator
- headContext = createRuntimeContext(configuration);
+ headContext = createRuntimeContext(configuration, accumulatorMap);
this.contexts.add(headContext);
streamOperator.setup(outputHandler.getOutput(), headContext);
}
@@ -105,14 +113,14 @@ public abstract class StreamTask<OUT, O extends StreamOperator<OUT>> extends Abs
return getEnvironment().getTaskName();
}
- public StreamingRuntimeContext createRuntimeContext(StreamConfig conf) {
+ public StreamingRuntimeContext createRuntimeContext(StreamConfig conf, Map<String, Accumulator<?,?>> accumulatorMap) {
Environment env = getEnvironment();
String operatorName = conf.getStreamOperator(userClassLoader).getClass().getSimpleName();
KeySelector<?,Serializable> statePartitioner = conf.getStatePartitioner(userClassLoader);
return new StreamingRuntimeContext(operatorName, env, getUserCodeClassLoader(),
- getExecutionConfig(), statePartitioner, getStateHandleProvider());
+ getExecutionConfig(), statePartitioner, getStateHandleProvider(), accumulatorMap);
}
private StateHandleProvider<Serializable> getStateHandleProvider() {
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamingRuntimeContext.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamingRuntimeContext.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamingRuntimeContext.java
index 3efd619..7eff16e 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamingRuntimeContext.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamingRuntimeContext.java
@@ -26,6 +26,7 @@ import java.util.List;
import java.util.Map;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.functions.util.RuntimeUDFContext;
import org.apache.flink.api.common.state.OperatorState;
@@ -55,9 +56,9 @@ public class StreamingRuntimeContext extends RuntimeUDFContext {
@SuppressWarnings("unchecked")
public StreamingRuntimeContext(String name, Environment env, ClassLoader userCodeClassLoader,
ExecutionConfig executionConfig, KeySelector<?, ?> statePartitioner,
- StateHandleProvider<?> provider) {
+ StateHandleProvider<?> provider, Map<String, Accumulator<?, ?>> accumulatorMap) {
super(name, env.getNumberOfSubtasks(), env.getIndexInSubtaskGroup(), userCodeClassLoader,
- executionConfig, env.getDistributedCacheEntries());
+ executionConfig, env.getDistributedCacheEntries(), accumulatorMap);
this.env = env;
this.statePartitioner = statePartitioner;
this.states = new HashMap<String, StreamOperatorState>();
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/state/StatefulOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/state/StatefulOperatorTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/state/StatefulOperatorTest.java
index 8c7ffeb..eb49e26 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/state/StatefulOperatorTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/state/StatefulOperatorTest.java
@@ -25,11 +25,13 @@ import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.OperatorState;
import org.apache.flink.api.java.functions.KeySelector;
@@ -138,7 +140,7 @@ public class StatefulOperatorTest {
StreamingRuntimeContext context = new StreamingRuntimeContext("MockTask", new MockEnvironment(3 * 1024 * 1024,
new MockInputSplitProvider(), 1024), null, new ExecutionConfig(), partitioner,
- new LocalStateHandleProvider<Serializable>());
+ new LocalStateHandleProvider<Serializable>(), new HashMap<String, Accumulator<?, ?>>());
StreamMap<Integer, String> op = new StreamMap<Integer, String>(new StatefulMapper());
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/io/BarrierBufferTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/io/BarrierBufferTest.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/io/BarrierBufferTest.java
index 0afe8b5..89ec7dc 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/io/BarrierBufferTest.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/io/BarrierBufferTest.java
@@ -24,7 +24,9 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
+import org.apache.flink.api.common.accumulators.LongCounter;
import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.event.task.TaskEvent;
import org.apache.flink.runtime.io.network.api.EndOfPartitionEvent;
import org.apache.flink.runtime.io.network.api.reader.AbstractReader;
@@ -198,6 +200,10 @@ public class BarrierBufferTest {
super(inputGate);
}
+ @Override
+ public void setReporter(AccumulatorRegistry.Reporter reporter) {
+
+ }
}
protected static BufferOrEvent createSuperstep(long id, int channel) {
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamMockEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamMockEnvironment.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamMockEnvironment.java
index 9864115..2092d83 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamMockEnvironment.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamMockEnvironment.java
@@ -23,6 +23,7 @@ import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.broadcast.BroadcastVariableManager;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
@@ -82,6 +83,8 @@ public class StreamMockEnvironment implements Environment {
private final BroadcastVariableManager bcVarManager = new BroadcastVariableManager();
+ private final AccumulatorRegistry accumulatorRegistry;
+
private final int bufferSize;
public StreamMockEnvironment(long memorySize, MockInputSplitProvider inputSplitProvider, int bufferSize) {
@@ -94,6 +97,8 @@ public class StreamMockEnvironment implements Environment {
this.ioManager = new IOManagerAsync();
this.inputSplitProvider = inputSplitProvider;
this.bufferSize = bufferSize;
+
+ this.accumulatorRegistry = new AccumulatorRegistry(jobID, getExecutionId());
}
public IteratorWrappingTestSingleInputGate<Record> addInput(MutableObjectIterator<Record> inputIterator) {
@@ -262,8 +267,8 @@ public class StreamMockEnvironment implements Environment {
}
@Override
- public void reportAccumulators(Map<String, Accumulator<?, ?>> accumulators) {
- // discard, this is only for testing
+ public AccumulatorRegistry getAccumulatorRegistry() {
+ return accumulatorRegistry;
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/MockCoContext.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/MockCoContext.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/MockCoContext.java
index 344fc7d..0467b5f 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/MockCoContext.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/MockCoContext.java
@@ -20,10 +20,12 @@ package org.apache.flink.streaming.util;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.typeutils.TypeExtractor;
@@ -157,7 +159,7 @@ public class MockCoContext<IN1, IN2, OUT> {
MockCoContext<IN1, IN2, OUT> mockContext = new MockCoContext<IN1, IN2, OUT>(input1, input2);
StreamingRuntimeContext runtimeContext = new StreamingRuntimeContext("CoMockTask",
new MockEnvironment(3 * 1024 * 1024, new MockInputSplitProvider(), 1024), null,
- new ExecutionConfig(), null, null);
+ new ExecutionConfig(), null, null, new HashMap<String, Accumulator<?, ?>>());
operator.setup(mockContext.collector, runtimeContext);
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/MockContext.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/MockContext.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/MockContext.java
index fc5079a..0d09c14 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/MockContext.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/MockContext.java
@@ -20,10 +20,12 @@ package org.apache.flink.streaming.util;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.runtime.operators.testutils.MockEnvironment;
@@ -105,7 +107,7 @@ public class MockContext<IN, OUT> {
MockContext<IN, OUT> mockContext = new MockContext<IN, OUT>(inputs);
StreamingRuntimeContext runtimeContext = new StreamingRuntimeContext("MockTask",
new MockEnvironment(3 * 1024 * 1024, new MockInputSplitProvider(), 1024), null,
- new ExecutionConfig(), null, null);
+ new ExecutionConfig(), null, null, new HashMap<String, Accumulator<?, ?>>());
operator.setup(mockContext.output, runtimeContext);
try {
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/SourceFunctionUtil.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/SourceFunctionUtil.java b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/SourceFunctionUtil.java
index dafd9a3..764fe5f 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/SourceFunctionUtil.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/SourceFunctionUtil.java
@@ -19,9 +19,11 @@ package org.apache.flink.streaming.util;
import java.io.Serializable;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.functions.RichFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
@@ -38,7 +40,7 @@ public class SourceFunctionUtil<T> {
List<T> outputs = new ArrayList<T>();
if (sourceFunction instanceof RichFunction) {
RuntimeContext runtimeContext = new StreamingRuntimeContext("MockTask", new MockEnvironment(3 * 1024 * 1024, new MockInputSplitProvider(), 1024), null,
- new ExecutionConfig(), null, new LocalStateHandleProvider<Serializable>());
+ new ExecutionConfig(), null, new LocalStateHandleProvider<Serializable>(), new HashMap<String, Accumulator<?, ?>>());
((RichFunction) sourceFunction).setRuntimeContext(runtimeContext);
((RichFunction) sourceFunction).open(new Configuration());
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-staging/flink-tez/src/main/java/org/apache/flink/tez/runtime/RegularProcessor.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-tez/src/main/java/org/apache/flink/tez/runtime/RegularProcessor.java b/flink-staging/flink-tez/src/main/java/org/apache/flink/tez/runtime/RegularProcessor.java
index 6b39734..5bfa49b 100644
--- a/flink-staging/flink-tez/src/main/java/org/apache/flink/tez/runtime/RegularProcessor.java
+++ b/flink-staging/flink-tez/src/main/java/org/apache/flink/tez/runtime/RegularProcessor.java
@@ -19,6 +19,7 @@
package org.apache.flink.tez.runtime;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.functions.Function;
import org.apache.flink.runtime.operators.PactDriver;
import org.apache.flink.api.common.functions.util.RuntimeUDFContext;
@@ -68,7 +69,8 @@ public class RegularProcessor<S extends Function, OT> extends AbstractLogicalIOP
getContext().getVertexParallelism(),
getContext().getTaskIndex(),
getClass().getClassLoader(),
- new ExecutionConfig());
+ new ExecutionConfig(),
+ new HashMap<String, Accumulator<?, ?>>());
this.task = new TezTask<S, OT>(taskConfig, runtimeUdfContext, this.getContext().getTotalMemoryAvailableToTask());
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-test-utils/src/main/java/org/apache/flink/test/util/RecordAPITestBase.java
----------------------------------------------------------------------
diff --git a/flink-test-utils/src/main/java/org/apache/flink/test/util/RecordAPITestBase.java b/flink-test-utils/src/main/java/org/apache/flink/test/util/RecordAPITestBase.java
index 1b39dbd..d23469e 100644
--- a/flink-test-utils/src/main/java/org/apache/flink/test/util/RecordAPITestBase.java
+++ b/flink-test-utils/src/main/java/org/apache/flink/test/util/RecordAPITestBase.java
@@ -70,7 +70,7 @@ public abstract class RecordAPITestBase extends AbstractTestBase {
Plan p = getTestJob();
p.setExecutionConfig(new ExecutionConfig());
if (p == null) {
- Assert.fail("Error: Cannot obtain Pact plan. Did the thest forget to override either 'getPactPlan()' or 'getJobGraph()' ?");
+ Assert.fail("Error: Cannot obtain Pact plan. Did the test forget to override either 'getPactPlan()' or 'getJobGraph()' ?");
}
Optimizer pc = new Optimizer(new DataStatistics(), this.config);
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
----------------------------------------------------------------------
diff --git a/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala b/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
index 0534178..509b86f 100644
--- a/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
+++ b/flink-test-utils/src/main/scala/org/apache/flink/test/util/ForkableFlinkMiniCluster.scala
@@ -85,7 +85,6 @@ class ForkableFlinkMiniCluster(userConfiguration: Configuration,
scheduler,
libraryCacheManager,
_,
- accumulatorManager,
executionRetries,
delayBetweenRetries,
timeout,
@@ -106,7 +105,6 @@ class ForkableFlinkMiniCluster(userConfiguration: Configuration,
scheduler,
libraryCacheManager,
archive,
- accumulatorManager,
executionRetries,
delayBetweenRetries,
timeout,
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorITCase.java b/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorITCase.java
index 11d6b83..d7d45fe 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorITCase.java
@@ -127,8 +127,7 @@ public class AccumulatorITCase extends JavaProgramTestBase {
// Add built-in accumulator without convenience function
getRuntimeContext().addAccumulator("open-close-counter", this.openCloseCounter);
- // Add custom counter. Didn't find a way to do this with
- // getAccumulator()
+ // Add custom counter
this.distinctWords = new SetAccumulator<StringValue>();
this.getRuntimeContext().addAccumulator("distinct-words", distinctWords);
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorIterativeITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorIterativeITCase.java b/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorIterativeITCase.java
index 78bbe68..6221c08 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorIterativeITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorIterativeITCase.java
@@ -72,7 +72,7 @@ public class AccumulatorIterativeITCase extends RecordAPITestBase {
protected void postSubmit() throws Exception {
compareResultsByLinesInMemory(EXPECTED, resultPath);
- Integer res = (Integer) getJobExecutionResult().getAccumulatorResult("test");
+ Integer res = getJobExecutionResult().getAccumulatorResult("test");
Assert.assertEquals(Integer.valueOf(NUM_ITERATIONS * 6), res);
}
@@ -117,18 +117,12 @@ public class AccumulatorIterativeITCase extends RecordAPITestBase {
private static final long serialVersionUID = 1L;
private IntCounter testCounter = new IntCounter();
-
- @Override
- public void open(Configuration parameters) throws Exception {
- super.open(parameters);
- getRuntimeContext().addAccumulator("test", this.testCounter);
- }
-
+
@Override
public void reduce(Iterator<Record> records, Collector<Record> out) {
// Compute the sum
int sum = 0;
-
+
while (records.hasNext()) {
Record r = records.next();
Integer value = Integer.parseInt(r.getField(0, StringValue.class).getValue());
@@ -137,6 +131,12 @@ public class AccumulatorIterativeITCase extends RecordAPITestBase {
}
out.collect(new Record(new StringValue(Integer.toString(sum))));
}
+
+ @Override
+ public void close() throws Exception {
+ super.close();
+ getRuntimeContext().addAccumulator("test", this.testCounter);
+ }
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorLiveITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorLiveITCase.java b/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorLiveITCase.java
new file mode 100644
index 0000000..84c50a9
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorLiveITCase.java
@@ -0,0 +1,299 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.test.accumulators;
+
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.actor.Status;
+import akka.testkit.JavaTestKit;
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.Plan;
+import org.apache.flink.api.common.accumulators.Accumulator;
+import org.apache.flink.api.common.accumulators.IntCounter;
+import org.apache.flink.api.common.accumulators.LongCounter;
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.RichFlatMapFunction;
+import org.apache.flink.api.common.io.OutputFormat;
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.LocalEnvironment;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.optimizer.DataStatistics;
+import org.apache.flink.optimizer.Optimizer;
+import org.apache.flink.optimizer.plan.OptimizedPlan;
+import org.apache.flink.optimizer.plantranslate.JobGraphGenerator;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
+import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.messages.JobManagerMessages;
+import org.apache.flink.runtime.taskmanager.TaskManager;
+import org.apache.flink.runtime.testingUtils.TestingCluster;
+import org.apache.flink.runtime.testingUtils.TestingJobManagerMessages.*;
+import org.apache.flink.runtime.testingUtils.TestingUtils;
+import org.apache.flink.util.Collector;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.*;
+
+
+/**
+ * Test the availability of accumulator results during runtime.
+ */
+@SuppressWarnings("serial")
+public class AccumulatorLiveITCase {
+
+ private static ActorSystem system;
+ private static ActorRef jobManager;
+
+ // name of accumulator
+ private static String NAME = "test";
+ // time to wait between changing the accumulator value
+ private static long WAIT_TIME = TaskManager.HEARTBEAT_INTERVAL().toMillis() + 500;
+
+ // number of heartbeat intervals to check
+ private static int NUM_ITERATIONS = 3;
+ // numer of retries in case the expected value is not seen
+ private static int NUM_RETRIES = 10;
+
+ private static List<String> inputData = new ArrayList<String>(NUM_ITERATIONS);
+
+
+ @Before
+ public void before() throws Exception {
+ system = AkkaUtils.createLocalActorSystem(new Configuration());
+ TestingCluster testingCluster = TestingUtils.startTestingCluster(1, 1, TestingUtils.DEFAULT_AKKA_ASK_TIMEOUT());
+ jobManager = testingCluster.getJobManager();
+
+ // generate test data
+ for (int i=0; i < NUM_ITERATIONS; i++) {
+ inputData.add(i, String.valueOf(i+1));
+ }
+ }
+
+ @After
+ public void after() throws Exception {
+ JavaTestKit.shutdownActorSystem(system);
+ }
+
+ @Test
+ public void testProgram() throws Exception {
+
+ new JavaTestKit(system) {{
+
+ /** The program **/
+ ExecutionEnvironment env = new PlanExtractor();
+ DataSet<String> input = env.fromCollection(inputData);
+ input
+ .flatMap(new Tokenizer())
+ .flatMap(new WaitingUDF())
+ .output(new WaitingOutputFormat());
+ env.execute();
+
+ /** Extract job graph **/
+ JobGraph jobGraph = getOptimizedPlan(((PlanExtractor) env).plan);
+
+ jobManager.tell(new JobManagerMessages.SubmitJob(jobGraph, false), getRef());
+ expectMsgClass(Status.Success.class);
+
+ /* Check for accumulator values */
+ int i = 0, retries = 0;
+ int expectedAccVal = 0;
+ while(i <= NUM_ITERATIONS) {
+ if (retries > 0) {
+ // retry fast
+ Thread.sleep(WAIT_TIME / NUM_RETRIES);
+ } else {
+ // wait for heartbeat interval
+ Thread.sleep(WAIT_TIME);
+ }
+
+ jobManager.tell(new RequestAccumulatorValues(jobGraph.getJobID()), getRef());
+ RequestAccumulatorValuesResponse response =
+ expectMsgClass(RequestAccumulatorValuesResponse.class);
+
+ Map<String, Accumulator<?, ?>> userAccumulators = response.userAccumulators();
+ Map<ExecutionAttemptID, Map<AccumulatorRegistry.Metric, Accumulator<?,?>>> flinkAccumulators =
+ response.flinkAccumulators();
+
+ if (checkUserAccumulators(expectedAccVal, userAccumulators) && checkFlinkAccumulators(i == NUM_ITERATIONS, i, i * 4, flinkAccumulators)) {
+// System.out.println("Passed round " + i);
+ // We passed this round
+ i += 1;
+ expectedAccVal += i;
+ retries = 0;
+ } else {
+ if (retries < NUM_RETRIES) {
+// System.out.println("retrying for the " + retries + " time.");
+ // try again
+ retries += 1;
+ } else {
+ fail("Failed in round #" + i + " after " + retries + " retries.");
+ }
+ }
+ }
+
+// expectMsgClass(new FiniteDuration(10, TimeUnit.SECONDS), JobManagerMessages.JobResultSuccess.class);
+ }};
+ }
+
+ private static boolean checkUserAccumulators(int expected, Map<String, Accumulator<?,?>> accumulatorMap) {
+// System.out.println("checking user accumulators");
+ return accumulatorMap.containsKey(NAME) && expected == ((IntCounter)accumulatorMap.get(NAME)).getLocalValue();
+ }
+
+ private static boolean checkFlinkAccumulators(boolean lastRound, int expectedRecords, int expectedBytes,
+ Map<ExecutionAttemptID, Map<AccumulatorRegistry.Metric, Accumulator<?,?>>> accumulatorMap) {
+// System.out.println("checking flink accumulators");
+ boolean returnValue = false;
+
+ for(Map<AccumulatorRegistry.Metric, Accumulator<?,?>> taskMap : accumulatorMap.values()) {
+ if (taskMap != null) {
+ for (Map.Entry<AccumulatorRegistry.Metric, Accumulator<?, ?>> entry : taskMap.entrySet()) {
+ switch (entry.getKey()) {
+ /**
+ * The following two cases are for the DataSource and Map task
+ */
+ case NUM_RECORDS_OUT:
+ if (!lastRound) {
+ assertTrue(((LongCounter) entry.getValue()).getLocalValue() == expectedRecords);
+ returnValue = true;
+ }
+ break;
+ case NUM_BYTES_OUT:
+ if (!lastRound) {
+ assertTrue(((LongCounter) entry.getValue()).getLocalValue() == expectedBytes);
+ returnValue = true;
+ }
+ break;
+ /**
+ * The following two cases are for the DataSink task
+ */
+ case NUM_RECORDS_IN:
+ // check if we are in last round and in current task accumulator map
+ if (lastRound && ((LongCounter)taskMap.get(AccumulatorRegistry.Metric.NUM_RECORDS_OUT)).getLocalValue() == 0) {
+ assertTrue(((LongCounter) entry.getValue()).getLocalValue() == expectedRecords);
+ returnValue = true;
+ }
+ break;
+ case NUM_BYTES_IN:
+ if (lastRound && ((LongCounter)taskMap.get(AccumulatorRegistry.Metric.NUM_RECORDS_OUT)).getLocalValue() == 0) {
+ assertTrue(((LongCounter) entry.getValue()).getLocalValue() == expectedBytes);
+ returnValue = true;
+ }
+ break;
+ default:
+ fail("Unknown accumulator found.");
+ }
+ }
+ }
+ }
+ return returnValue;
+ }
+
+
+ public static class Tokenizer implements FlatMapFunction<String, String> {
+
+ @Override
+ public void flatMap(String value, Collector<String> out) throws Exception {
+ for (String str : value.split("\n")) {
+ out.collect(str);
+ }
+ }
+ }
+
+ /**
+ * UDF that waits for at least the heartbeat interval's duration.
+ */
+ private static class WaitingUDF extends RichFlatMapFunction<String, Integer> {
+
+ private IntCounter counter = new IntCounter();
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ getRuntimeContext().addAccumulator(NAME, counter);
+ }
+
+ @Override
+ public void flatMap(String value, Collector<Integer> out) throws Exception {
+ /* Wait here to check the accumulator value in the meantime */
+ Thread.sleep(WAIT_TIME);
+ int val = Integer.valueOf(value);
+ counter.add(val);
+ out.collect(val);
+ }
+ }
+
+ private static class WaitingOutputFormat implements OutputFormat<Integer> {
+
+ @Override
+ public void configure(Configuration parameters) {
+
+ }
+
+ @Override
+ public void open(int taskNumber, int numTasks) throws IOException {
+
+ }
+
+ @Override
+ public void writeRecord(Integer record) throws IOException {
+ }
+
+ @Override
+ public void close() throws IOException {
+ try {
+// System.out.println("starting output task");
+ Thread.sleep(WAIT_TIME);
+ } catch (InterruptedException e) {
+ fail("Interrupted test.");
+ }
+ }
+ }
+
+ /**
+ * Helpers to generate the JobGraph
+ */
+ private static JobGraph getOptimizedPlan(Plan plan) {
+ Optimizer pc = new Optimizer(new DataStatistics(), new Configuration());
+ JobGraphGenerator jgg = new JobGraphGenerator();
+ OptimizedPlan op = pc.compile(plan);
+ return jgg.compileJobGraph(op);
+ }
+
+ private static class PlanExtractor extends LocalEnvironment {
+
+ private Plan plan = null;
+
+ @Override
+ public JobExecutionResult execute(String jobName) throws Exception {
+ plan = createProgramPlan();
+ return new JobExecutionResult(new JobID(), -1, null);
+ }
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationMaster.scala
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationMaster.scala b/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationMaster.scala
index 8cfeead..c497a90 100644
--- a/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationMaster.scala
+++ b/flink-yarn/src/main/scala/org/apache/flink/yarn/ApplicationMaster.scala
@@ -240,7 +240,6 @@ object ApplicationMaster {
scheduler,
libraryCacheManager,
archiveProps,
- accumulatorManager,
executionRetries,
delayBetweenRetries,
timeout,
@@ -257,7 +256,6 @@ object ApplicationMaster {
scheduler,
libraryCacheManager,
archiver,
- accumulatorManager,
executionRetries,
delayBetweenRetries,
timeout,
[3/4] flink git commit: [FLINK-2292][FLINK-1573] add live per-task
accumulators
Posted by mx...@apache.org.
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java
index 4381fd0..29efc4c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.io.network.api.writer;
import org.apache.flink.core.io.IOReadableWritable;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.event.task.AbstractEvent;
import org.apache.flink.runtime.io.network.api.serialization.RecordSerializer;
import org.apache.flink.runtime.io.network.api.serialization.SpanningRecordSerializer;
@@ -49,6 +50,12 @@ public class RecordWriter<T extends IOReadableWritable> {
private final int numChannels;
+ /**
+ * Counter for the number of records emitted and for the number of bytes written.
+ * @param counter
+ */
+ private AccumulatorRegistry.Reporter reporter;
+
/** {@link RecordSerializer} per outgoing channel */
private final RecordSerializer<T>[] serializers;
@@ -81,6 +88,7 @@ public class RecordWriter<T extends IOReadableWritable> {
synchronized (serializer) {
SerializationResult result = serializer.addRecord(record);
+
while (result.isFullBuffer()) {
Buffer buffer = serializer.getCurrentBuffer();
@@ -90,8 +98,18 @@ public class RecordWriter<T extends IOReadableWritable> {
}
buffer = writer.getBufferProvider().requestBufferBlocking();
+ if (reporter != null) {
+ // increase the number of written bytes by the memory segment's size
+ reporter.reportNumBytesOut(buffer.getSize());
+ }
+
result = serializer.setNextBuffer(buffer);
}
+
+ if(reporter != null) {
+ // count number of emitted records
+ reporter.reportNumRecordsOut(1);
+ }
}
}
}
@@ -173,4 +191,14 @@ public class RecordWriter<T extends IOReadableWritable> {
}
}
}
+
+ /**
+ * Counter for the number of records emitted and the records processed.
+ */
+ public void setReporter(AccumulatorRegistry.Reporter reporter) {
+ for(RecordSerializer<?> serializer : serializers) {
+ serializer.setReporter(reporter);
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/AbstractIterativePactTask.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/AbstractIterativePactTask.java b/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/AbstractIterativePactTask.java
index 9c5fdca..72434e6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/AbstractIterativePactTask.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/AbstractIterativePactTask.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.iterative.task;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.accumulators.Accumulator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.flink.api.common.aggregators.Aggregator;
@@ -52,6 +53,8 @@ import org.apache.flink.util.InstantiationUtil;
import org.apache.flink.util.MutableObjectIterator;
import java.io.IOException;
+import java.io.Serializable;
+import java.util.Map;
/**
* The abstract base class for all tasks able to participate in an iteration.
@@ -166,7 +169,7 @@ public abstract class AbstractIterativePactTask<S extends Function, OT> extends
public DistributedRuntimeUDFContext createRuntimeContext(String taskName) {
Environment env = getEnvironment();
return new IterativeRuntimeUdfContext(taskName, env.getNumberOfSubtasks(),
- env.getIndexInSubtaskGroup(), getUserCodeClassLoader(), getExecutionConfig());
+ env.getIndexInSubtaskGroup(), getUserCodeClassLoader(), getExecutionConfig(), this.accumulatorMap);
}
// --------------------------------------------------------------------------------------------
@@ -356,8 +359,10 @@ public abstract class AbstractIterativePactTask<S extends Function, OT> extends
private class IterativeRuntimeUdfContext extends DistributedRuntimeUDFContext implements IterationRuntimeContext {
- public IterativeRuntimeUdfContext(String name, int numParallelSubtasks, int subtaskIndex, ClassLoader userCodeClassLoader, ExecutionConfig executionConfig) {
- super(name, numParallelSubtasks, subtaskIndex, userCodeClassLoader, executionConfig);
+ public IterativeRuntimeUdfContext(String name, int numParallelSubtasks, int subtaskIndex, ClassLoader userCodeClassLoader,
+ ExecutionConfig executionConfig,
+ Map<String, Accumulator<?,?>> accumulatorMap) {
+ super(name, numParallelSubtasks, subtaskIndex, userCodeClassLoader, executionConfig, accumulatorMap);
}
@Override
@@ -375,6 +380,14 @@ public abstract class AbstractIterativePactTask<S extends Function, OT> extends
public <T extends Value> T getPreviousIterationAggregate(String name) {
return (T) getIterationAggregators().getPreviousGlobalAggregate(name);
}
+
+ @Override
+ public <V, A extends Serializable> void addAccumulator(String name, Accumulator<V, A> newAccumulator) {
+ // only add accumulator on first iteration
+ if (inFirstIteration()) {
+ super.addAccumulator(name, newAccumulator);
+ }
+ }
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/IterationHeadPactTask.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/IterationHeadPactTask.java b/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/IterationHeadPactTask.java
index cf02bdf..9cb045f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/IterationHeadPactTask.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/iterative/task/IterationHeadPactTask.java
@@ -22,6 +22,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.io.network.api.writer.ResultPartitionWriter;
import org.apache.flink.runtime.io.network.api.writer.RecordWriter;
import org.slf4j.Logger;
@@ -112,8 +113,9 @@ public class IterationHeadPactTask<X, Y, S extends Function, OT> extends Abstrac
List<RecordWriter<?>> finalOutputWriters = new ArrayList<RecordWriter<?>>();
final TaskConfig finalOutConfig = this.config.getIterationHeadFinalOutputConfig();
final ClassLoader userCodeClassLoader = getUserCodeClassLoader();
+ AccumulatorRegistry.Reporter reporter = getEnvironment().getAccumulatorRegistry().getReadWriteReporter();
this.finalOutputCollector = RegularPactTask.getOutputCollector(this, finalOutConfig,
- userCodeClassLoader, finalOutputWriters, config.getNumOutputs(), finalOutConfig.getNumOutputs());
+ userCodeClassLoader, finalOutputWriters, config.getNumOutputs(), finalOutConfig.getNumOutputs(), reporter);
// sanity check the setup
final int writersIntoStepFunction = this.eventualOutputs.size();
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/AbstractInvokable.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/AbstractInvokable.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/AbstractInvokable.java
index 85dd5c5..df41672 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/AbstractInvokable.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/tasks/AbstractInvokable.java
@@ -26,7 +26,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * This is the abstract base class for every task that can be executed ba a TaskManager.
+ * This is the abstract base class for every task that can be executed by a TaskManager.
* Concrete tasks like the stream vertices of the batch tasks
* (see {@link org.apache.flink.runtime.operators.RegularPactTask}) inherit from this class.
*
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/accumulators/AccumulatorManager.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/accumulators/AccumulatorManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/accumulators/AccumulatorManager.java
deleted file mode 100644
index c824232..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/accumulators/AccumulatorManager.java
+++ /dev/null
@@ -1,144 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.jobmanager.accumulators;
-
-import java.io.IOException;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.Map;
-
-import org.apache.flink.api.common.accumulators.Accumulator;
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
-import org.apache.flink.runtime.util.SerializedValue;
-
-/**
- * This class manages the accumulators for different jobs. Either the jobs are
- * running and new accumulator results have to be merged in, or the jobs are no
- * longer running and the results shall be still available for the client or the
- * web interface. Accumulators for older jobs are automatically removed when new
- * arrive, based on a maximum number of entries.
- *
- * All functions are thread-safe and thus can be called directly from
- * JobManager.
- */
-public class AccumulatorManager {
-
- /** Map of accumulators belonging to recently started jobs */
- private final Map<JobID, JobAccumulators> jobAccumulators = new HashMap<JobID, JobAccumulators>();
-
- private final LinkedList<JobID> lru = new LinkedList<JobID>();
- private int maxEntries;
-
-
- public AccumulatorManager(int maxEntries) {
- this.maxEntries = maxEntries;
- }
-
- /**
- * Merges the new accumulators with the existing accumulators collected for
- * the job.
- */
- public void processIncomingAccumulators(JobID jobID,
- Map<String, Accumulator<?, ?>> newAccumulators) {
- synchronized (this.jobAccumulators) {
-
- JobAccumulators jobAccumulators = this.jobAccumulators.get(jobID);
- if (jobAccumulators == null) {
- jobAccumulators = new JobAccumulators();
- this.jobAccumulators.put(jobID, jobAccumulators);
- cleanup(jobID);
- }
- jobAccumulators.processNew(newAccumulators);
- }
- }
-
- public Map<String, Object> getJobAccumulatorResults(JobID jobID) {
- Map<String, Object> result = new HashMap<String, Object>();
-
- JobAccumulators acc;
- synchronized (jobAccumulators) {
- acc = jobAccumulators.get(jobID);
- }
-
- if (acc != null) {
- for (Map.Entry<String, Accumulator<?, ?>> entry : acc.getAccumulators().entrySet()) {
- result.put(entry.getKey(), entry.getValue().getLocalValue());
- }
- }
-
- return result;
- }
-
- public Map<String, SerializedValue<Object>> getJobAccumulatorResultsSerialized(JobID jobID) throws IOException {
- JobAccumulators acc;
- synchronized (jobAccumulators) {
- acc = jobAccumulators.get(jobID);
- }
-
- if (acc == null || acc.getAccumulators().isEmpty()) {
- return Collections.emptyMap();
- }
-
- Map<String, SerializedValue<Object>> result = new HashMap<String, SerializedValue<Object>>();
- for (Map.Entry<String, Accumulator<?, ?>> entry : acc.getAccumulators().entrySet()) {
- result.put(entry.getKey(), new SerializedValue<Object>(entry.getValue().getLocalValue()));
- }
-
- return result;
- }
-
- public StringifiedAccumulatorResult[] getJobAccumulatorResultsStringified(JobID jobID) throws IOException {
- JobAccumulators acc;
- synchronized (jobAccumulators) {
- acc = jobAccumulators.get(jobID);
- }
-
- if (acc == null || acc.getAccumulators().isEmpty()) {
- return new StringifiedAccumulatorResult[0];
- }
-
- Map<String, Accumulator<?, ?>> accMap = acc.getAccumulators();
-
- StringifiedAccumulatorResult[] result = new StringifiedAccumulatorResult[accMap.size()];
- int i = 0;
- for (Map.Entry<String, Accumulator<?, ?>> entry : accMap.entrySet()) {
- String type = entry.getValue() == null ? "(null)" : entry.getValue().getClass().getSimpleName();
- String value = entry.getValue() == null ? "(null)" : entry.getValue().toString();
- result[i++] = new StringifiedAccumulatorResult(entry.getKey(), type, value);
- }
- return result;
- }
-
- /**
- * Cleanup data for the oldest jobs if the maximum number of entries is reached.
- *
- * @param jobId The (potentially new) JobId.
- */
- private void cleanup(JobID jobId) {
- if (!lru.contains(jobId)) {
- lru.addFirst(jobId);
- }
- if (lru.size() > this.maxEntries) {
- JobID toRemove = lru.removeLast();
- this.jobAccumulators.remove(toRemove);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/accumulators/JobAccumulators.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/accumulators/JobAccumulators.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/accumulators/JobAccumulators.java
deleted file mode 100644
index 970d993..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/accumulators/JobAccumulators.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.jobmanager.accumulators;
-
-import java.util.HashMap;
-import java.util.Map;
-
-import org.apache.flink.api.common.accumulators.Accumulator;
-import org.apache.flink.api.common.accumulators.AccumulatorHelper;
-
-/**
- * Simple class wrapping a map of accumulators for a single job. Just for better
- * handling.
- */
-public class JobAccumulators {
-
- private final Map<String, Accumulator<?, ?>> accumulators = new HashMap<String, Accumulator<?, ?>>();
-
- public Map<String, Accumulator<?, ?>> getAccumulators() {
- return this.accumulators;
- }
-
- public void processNew(Map<String, Accumulator<?, ?>> newAccumulators) {
- AccumulatorHelper.mergeInto(this.accumulators, newAccumulators);
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSinkTask.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSinkTask.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSinkTask.java
index 345e1ab..b3130a1 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSinkTask.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSinkTask.java
@@ -26,6 +26,7 @@ import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.TypeSerializerFactory;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.io.IOReadableWritable;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.execution.CancelTaskException;
import org.apache.flink.runtime.io.network.api.reader.MutableReader;
import org.apache.flink.runtime.io.network.api.reader.MutableRecordReader;
@@ -356,6 +357,11 @@ public class DataSinkTask<IT> extends AbstractInvokable {
throw new Exception("Illegal input group size in task configuration: " + groupSize);
}
+ final AccumulatorRegistry accumulatorRegistry = getEnvironment().getAccumulatorRegistry();
+ final AccumulatorRegistry.Reporter reporter = accumulatorRegistry.getReadWriteReporter();
+
+ inputReader.setReporter(reporter);
+
this.inputTypeSerializerFactory = this.config.getInputSerializer(0, getUserCodeClassLoader());
@SuppressWarnings({ "rawtypes" })
final MutableObjectIterator<?> iter = new ReaderIterator(inputReader, this.inputTypeSerializerFactory.getSerializer());
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSourceTask.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSourceTask.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSourceTask.java
index 0bbe4bf..3f1c642 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSourceTask.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSourceTask.java
@@ -25,6 +25,7 @@ import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.TypeSerializerFactory;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.io.InputSplit;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.execution.CancelTaskException;
import org.apache.flink.runtime.io.network.api.writer.RecordWriter;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
@@ -39,9 +40,9 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
+import java.util.Map;
import java.util.NoSuchElementException;
/**
@@ -187,25 +188,22 @@ public class DataSourceTask<OT> extends AbstractInvokable {
format.close();
}
} // end for all input splits
-
+
// close the collector. if it is a chaining task collector, it will close its chained tasks
this.output.close();
-
+
// close all chained tasks letting them report failure
RegularPactTask.closeChainedTasks(this.chainedTasks, this);
-
- // Merge and report accumulators
- RegularPactTask.reportAndClearAccumulators(getEnvironment(),
- new HashMap<String, Accumulator<?,?>>(), chainedTasks);
+
}
catch (Exception ex) {
// close the input, but do not report any exceptions, since we already have another root cause
try {
this.format.close();
} catch (Throwable ignored) {}
-
+
RegularPactTask.cancelChainedTasks(this.chainedTasks);
-
+
ex = ExceptionInChainedStubException.exceptionUnwrap(ex);
if (ex instanceof CancelTaskException) {
@@ -275,7 +273,7 @@ public class DataSourceTask<OT> extends AbstractInvokable {
catch (Throwable t) {
throw new RuntimeException("The user defined 'configure()' method caused an error: " + t.getMessage(), t);
}
-
+
// get the factory for the type serializer
this.serializerFactory = this.config.getOutputSerializer(userCodeClassLoader);
}
@@ -287,7 +285,14 @@ public class DataSourceTask<OT> extends AbstractInvokable {
private void initOutputs(ClassLoader cl) throws Exception {
this.chainedTasks = new ArrayList<ChainedDriver<?, ?>>();
this.eventualOutputs = new ArrayList<RecordWriter<?>>();
- this.output = RegularPactTask.initOutputs(this, cl, this.config, this.chainedTasks, this.eventualOutputs, getExecutionConfig());
+
+ final AccumulatorRegistry accumulatorRegistry = getEnvironment().getAccumulatorRegistry();
+ final AccumulatorRegistry.Reporter reporter = accumulatorRegistry.getReadWriteReporter();
+
+ Map<String, Accumulator<?, ?>> accumulatorMap = accumulatorRegistry.getUserMap();
+
+ this.output = RegularPactTask.initOutputs(this, cl, this.config, this.chainedTasks, this.eventualOutputs,
+ getExecutionConfig(), reporter, accumulatorMap);
}
// ------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/operators/PactDriver.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/PactDriver.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/PactDriver.java
index d0f4116..a53f5bf 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/PactDriver.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/PactDriver.java
@@ -74,7 +74,7 @@ public interface PactDriver<S extends Function, OT> {
* code typically signal situations where this instance in unable to proceed, exceptions
* from the user code should be forwarded.
*/
- void run() throws Exception;
+ void run() throws Exception;
/**
* This method is invoked in any case (clean termination and exception) at the end of the tasks operation.
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/operators/PactTaskContext.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/PactTaskContext.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/PactTaskContext.java
index b296506..bc23fa3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/PactTaskContext.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/PactTaskContext.java
@@ -50,7 +50,7 @@ public interface PactTaskContext<S, OT> {
MemoryManager getMemoryManager();
IOManager getIOManager();
-
+
<X> MutableObjectIterator<X> getInput(int index);
<X> TypeSerializerFactory<X> getInputSerializer(int index);
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/operators/RegularPactTask.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/RegularPactTask.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/RegularPactTask.java
index 1c3328e..78bf383 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/RegularPactTask.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/RegularPactTask.java
@@ -20,18 +20,17 @@ package org.apache.flink.runtime.operators;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.accumulators.Accumulator;
-import org.apache.flink.api.common.accumulators.AccumulatorHelper;
import org.apache.flink.api.common.distributions.DataDistribution;
import org.apache.flink.api.common.functions.GroupCombineFunction;
import org.apache.flink.api.common.functions.Function;
import org.apache.flink.api.common.functions.Partitioner;
-import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.functions.util.FunctionUtils;
import org.apache.flink.api.common.typeutils.TypeComparator;
import org.apache.flink.api.common.typeutils.TypeComparatorFactory;
import org.apache.flink.api.common.typeutils.TypeSerializerFactory;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.io.IOReadableWritable;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.broadcast.BroadcastVariableMaterialization;
import org.apache.flink.runtime.execution.CancelTaskException;
import org.apache.flink.runtime.execution.Environment;
@@ -71,7 +70,6 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -114,7 +112,7 @@ public class RegularPactTask<S extends Function, OT> extends AbstractInvokable i
protected List<RecordWriter<?>> eventualOutputs;
/**
- * The input readers to this task.
+ * The input readers of this task.
*/
protected MutableReader<?>[] inputReaders;
@@ -212,7 +210,11 @@ public class RegularPactTask<S extends Function, OT> extends AbstractInvokable i
*/
protected volatile boolean running = true;
-
+ /**
+ * The accumulator map used in the RuntimeContext.
+ */
+ protected Map<String, Accumulator<?,?>> accumulatorMap;
+
// --------------------------------------------------------------------------------------------
// Task Interface
// --------------------------------------------------------------------------------------------
@@ -273,7 +275,9 @@ public class RegularPactTask<S extends Function, OT> extends AbstractInvokable i
LOG.debug(formatLogString("Start task code."));
}
- this.runtimeUdfContext = createRuntimeContext(getEnvironment().getTaskName());
+ Environment env = getEnvironment();
+
+ this.runtimeUdfContext = createRuntimeContext(env.getTaskName());
// whatever happens in this scope, make sure that the local strategies are cleaned up!
// note that the initialization of the local strategies is in the try-finally block as well,
@@ -367,6 +371,7 @@ public class RegularPactTask<S extends Function, OT> extends AbstractInvokable i
clearReaders(inputReaders);
clearWriters(eventualOutputs);
+
}
if (this.running) {
@@ -505,18 +510,6 @@ public class RegularPactTask<S extends Function, OT> extends AbstractInvokable i
// close all chained tasks letting them report failure
RegularPactTask.closeChainedTasks(this.chainedTasks, this);
-
- // Collect the accumulators of all involved UDFs and send them to the
- // JobManager. close() has been called earlier for all involved UDFs
- // (using this.stub.close() and closeChainedTasks()), so UDFs can no longer
- // modify accumulators;
-
- // collect the counters from the udf in the core driver
- Map<String, Accumulator<?, ?>> accumulators =
- FunctionUtils.getFunctionRuntimeContext(this.stub, this.runtimeUdfContext).getAllAccumulators();
-
- // collect accumulators from chained tasks and report them
- reportAndClearAccumulators(getEnvironment(), accumulators, this.chainedTasks);
}
catch (Exception ex) {
// close the input, but do not report any exceptions, since we already have another root cause
@@ -557,60 +550,6 @@ public class RegularPactTask<S extends Function, OT> extends AbstractInvokable i
}
}
- /**
- * This method is called at the end of a task, receiving the accumulators of
- * the task and the chained tasks. It merges them into a single map of
- * accumulators and sends them to the JobManager.
- *
- * @param chainedTasks
- * Each chained task might have accumulators which will be merged
- * with the accumulators of the stub.
- */
- protected static void reportAndClearAccumulators(Environment env,
- Map<String, Accumulator<?, ?>> accumulators,
- ArrayList<ChainedDriver<?, ?>> chainedTasks) {
-
- // We can merge here the accumulators from the stub and the chained
- // tasks. Type conflicts can occur here if counters with same name but
- // different type were used.
-
- if (!chainedTasks.isEmpty()) {
- if (accumulators == null) {
- accumulators = new HashMap<String, Accumulator<?, ?>>();
- }
-
- for (ChainedDriver<?, ?> chainedTask : chainedTasks) {
- RuntimeContext rc = FunctionUtils.getFunctionRuntimeContext(chainedTask.getStub(), null);
- if (rc != null) {
- Map<String, Accumulator<?, ?>> chainedAccumulators = rc.getAllAccumulators();
- if (chainedAccumulators != null) {
- AccumulatorHelper.mergeInto(accumulators, chainedAccumulators);
- }
- }
- }
- }
-
- // Don't report if the UDF didn't collect any accumulators
- if (accumulators == null || accumulators.size() == 0) {
- return;
- }
-
- // Report accumulators to JobManager
- env.reportAccumulators(accumulators);
-
- // We also clear the accumulators, since stub instances might be reused
- // (e.g. in iterations) and we don't want to count twice. This may not be
- // done before sending
- AccumulatorHelper.resetAndClearAccumulators(accumulators);
-
- for (ChainedDriver<?, ?> chainedTask : chainedTasks) {
- RuntimeContext rc = FunctionUtils.getFunctionRuntimeContext(chainedTask.getStub(), null);
- if (rc != null) {
- AccumulatorHelper.resetAndClearAccumulators(rc.getAllAccumulators());
- }
- }
- }
-
protected void closeLocalStrategiesAndCaches() {
// make sure that all broadcast variable references held by this task are released
@@ -725,6 +664,9 @@ public class RegularPactTask<S extends Function, OT> extends AbstractInvokable i
int currentReaderOffset = 0;
+ AccumulatorRegistry registry = getEnvironment().getAccumulatorRegistry();
+ AccumulatorRegistry.Reporter reporter = registry.getReadWriteReporter();
+
for (int i = 0; i < numInputs; i++) {
// ---------------- create the input readers ---------------------
// in case where a logical input unions multiple physical inputs, create a union reader
@@ -744,6 +686,8 @@ public class RegularPactTask<S extends Function, OT> extends AbstractInvokable i
throw new Exception("Illegal input group size in task configuration: " + groupSize);
}
+ inputReaders[i].setReporter(reporter);
+
currentReaderOffset += groupSize;
}
this.inputReaders = inputReaders;
@@ -1073,14 +1017,21 @@ public class RegularPactTask<S extends Function, OT> extends AbstractInvokable i
ClassLoader userCodeClassLoader = getUserCodeClassLoader();
- this.output = initOutputs(this, userCodeClassLoader, this.config, this.chainedTasks, this.eventualOutputs, this.getExecutionConfig());
+ AccumulatorRegistry accumulatorRegistry = getEnvironment().getAccumulatorRegistry();
+ AccumulatorRegistry.Reporter reporter = accumulatorRegistry.getReadWriteReporter();
+
+ this.accumulatorMap = accumulatorRegistry.getUserMap();
+
+ this.output = initOutputs(this, userCodeClassLoader, this.config, this.chainedTasks, this.eventualOutputs,
+ this.getExecutionConfig(), reporter, this.accumulatorMap);
}
public DistributedRuntimeUDFContext createRuntimeContext(String taskName) {
Environment env = getEnvironment();
+
return new DistributedRuntimeUDFContext(taskName, env.getNumberOfSubtasks(),
env.getIndexInSubtaskGroup(), getUserCodeClassLoader(), getExecutionConfig(),
- env.getDistributedCacheEntries());
+ env.getDistributedCacheEntries(), this.accumulatorMap);
}
// --------------------------------------------------------------------------------------------
@@ -1257,7 +1208,7 @@ public class RegularPactTask<S extends Function, OT> extends AbstractInvokable i
* @return The OutputCollector that data produced in this task is submitted to.
*/
public static <T> Collector<T> getOutputCollector(AbstractInvokable task, TaskConfig config, ClassLoader cl,
- List<RecordWriter<?>> eventualOutputs, int outputOffset, int numOutputs) throws Exception
+ List<RecordWriter<?>> eventualOutputs, int outputOffset, int numOutputs, AccumulatorRegistry.Reporter reporter) throws Exception
{
if (numOutputs == 0) {
return null;
@@ -1286,11 +1237,15 @@ public class RegularPactTask<S extends Function, OT> extends AbstractInvokable i
}
final DataDistribution distribution = config.getOutputDataDistribution(i, cl);
final Partitioner<?> partitioner = config.getOutputPartitioner(i, cl);
-
+
oe = new RecordOutputEmitter(strategy, comparator, partitioner, distribution);
}
- writers.add(new RecordWriter<Record>(task.getEnvironment().getWriter(outputOffset + i), oe));
+ // setup accumulator counters
+ final RecordWriter<Record> recordWriter = new RecordWriter<Record>(task.getEnvironment().getWriter(outputOffset + i), oe);
+ recordWriter.setReporter(reporter);
+
+ writers.add(recordWriter);
}
if (eventualOutputs != null) {
eventualOutputs.addAll(writers);
@@ -1318,12 +1273,18 @@ public class RegularPactTask<S extends Function, OT> extends AbstractInvokable i
else {
final DataDistribution dataDist = config.getOutputDataDistribution(i, cl);
final Partitioner<?> partitioner = config.getOutputPartitioner(i, cl);
-
+
final TypeComparator<T> comparator = compFactory.createComparator();
oe = new OutputEmitter<T>(strategy, comparator, partitioner, dataDist);
}
- writers.add(new RecordWriter<SerializationDelegate<T>>(task.getEnvironment().getWriter(outputOffset + i), oe));
+ final RecordWriter<SerializationDelegate<T>> recordWriter =
+ new RecordWriter<SerializationDelegate<T>>(task.getEnvironment().getWriter(outputOffset + i), oe);
+
+ // setup live accumulator counters
+ recordWriter.setReporter(reporter);
+
+ writers.add(recordWriter);
}
if (eventualOutputs != null) {
eventualOutputs.addAll(writers);
@@ -1338,7 +1299,11 @@ public class RegularPactTask<S extends Function, OT> extends AbstractInvokable i
*/
@SuppressWarnings("unchecked")
public static <T> Collector<T> initOutputs(AbstractInvokable nepheleTask, ClassLoader cl, TaskConfig config,
- List<ChainedDriver<?, ?>> chainedTasksTarget, List<RecordWriter<?>> eventualOutputs, ExecutionConfig executionConfig)
+ List<ChainedDriver<?, ?>> chainedTasksTarget,
+ List<RecordWriter<?>> eventualOutputs,
+ ExecutionConfig executionConfig,
+ AccumulatorRegistry.Reporter reporter,
+ Map<String, Accumulator<?,?>> accumulatorMap)
throws Exception
{
final int numOutputs = config.getNumOutputs();
@@ -1370,12 +1335,12 @@ public class RegularPactTask<S extends Function, OT> extends AbstractInvokable i
final TaskConfig chainedStubConf = config.getChainedStubConfig(i);
final String taskName = config.getChainedTaskName(i);
- if (i == numChained -1) {
+ if (i == numChained - 1) {
// last in chain, instantiate the output collector for this task
- previous = getOutputCollector(nepheleTask, chainedStubConf, cl, eventualOutputs, 0, chainedStubConf.getNumOutputs());
+ previous = getOutputCollector(nepheleTask, chainedStubConf, cl, eventualOutputs, 0, chainedStubConf.getNumOutputs(), reporter);
}
- ct.setup(chainedStubConf, taskName, previous, nepheleTask, cl, executionConfig);
+ ct.setup(chainedStubConf, taskName, previous, nepheleTask, cl, executionConfig, accumulatorMap);
chainedTasksTarget.add(0, ct);
previous = ct;
@@ -1386,7 +1351,7 @@ public class RegularPactTask<S extends Function, OT> extends AbstractInvokable i
// else
// instantiate the output collector the default way from this configuration
- return getOutputCollector(nepheleTask , config, cl, eventualOutputs, 0, numOutputs);
+ return getOutputCollector(nepheleTask , config, cl, eventualOutputs, 0, numOutputs, reporter);
}
// --------------------------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/operators/chaining/ChainedDriver.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/chaining/ChainedDriver.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/chaining/ChainedDriver.java
index b4cfa27..ea6cfe3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/chaining/ChainedDriver.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/chaining/ChainedDriver.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.operators.chaining;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.functions.Function;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.runtime.execution.Environment;
@@ -28,6 +29,8 @@ import org.apache.flink.runtime.operators.util.DistributedRuntimeUDFContext;
import org.apache.flink.runtime.operators.util.TaskConfig;
import org.apache.flink.util.Collector;
+import java.util.Map;
+
/**
* The interface to be implemented by drivers that do not run in an own pact task context, but are chained to other
* tasks.
@@ -50,20 +53,23 @@ public abstract class ChainedDriver<IT, OT> implements Collector<IT> {
public void setup(TaskConfig config, String taskName, Collector<OT> outputCollector,
- AbstractInvokable parent, ClassLoader userCodeClassLoader, ExecutionConfig executionConfig)
+ AbstractInvokable parent, ClassLoader userCodeClassLoader, ExecutionConfig executionConfig,
+ Map<String, Accumulator<?,?>> accumulatorMap)
{
this.config = config;
this.taskName = taskName;
this.outputCollector = outputCollector;
this.userCodeClassLoader = userCodeClassLoader;
-
+
+ Environment env = parent.getEnvironment();
+
if (parent instanceof RegularPactTask) {
this.udfContext = ((RegularPactTask<?, ?>) parent).createRuntimeContext(taskName);
} else {
- Environment env = parent.getEnvironment();
this.udfContext = new DistributedRuntimeUDFContext(taskName, env.getNumberOfSubtasks(),
env.getIndexInSubtaskGroup(), userCodeClassLoader, parent.getExecutionConfig(),
- env.getDistributedCacheEntries());
+ env.getDistributedCacheEntries(), accumulatorMap
+ );
}
this.executionConfig = executionConfig;
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/operators/util/DistributedRuntimeUDFContext.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/util/DistributedRuntimeUDFContext.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/util/DistributedRuntimeUDFContext.java
index f4cd354..4b480ba 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/util/DistributedRuntimeUDFContext.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/util/DistributedRuntimeUDFContext.java
@@ -24,6 +24,7 @@ import java.util.Map;
import java.util.concurrent.Future;
import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.functions.BroadcastVariableInitializer;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.functions.util.AbstractRuntimeUDFContext;
@@ -41,12 +42,14 @@ public class DistributedRuntimeUDFContext extends AbstractRuntimeUDFContext {
private final HashMap<String, BroadcastVariableMaterialization<?, ?>> broadcastVars = new HashMap<String, BroadcastVariableMaterialization<?, ?>>();
- public DistributedRuntimeUDFContext(String name, int numParallelSubtasks, int subtaskIndex, ClassLoader userCodeClassLoader, ExecutionConfig executionConfig) {
- super(name, numParallelSubtasks, subtaskIndex, userCodeClassLoader, executionConfig);
+ public DistributedRuntimeUDFContext(String name, int numParallelSubtasks, int subtaskIndex, ClassLoader userCodeClassLoader,
+ ExecutionConfig executionConfig, Map<String, Accumulator<?,?>> accumulators) {
+ super(name, numParallelSubtasks, subtaskIndex, userCodeClassLoader, executionConfig, accumulators);
}
- public DistributedRuntimeUDFContext(String name, int numParallelSubtasks, int subtaskIndex, ClassLoader userCodeClassLoader, ExecutionConfig executionConfig, Map<String, Future<Path>> cpTasks) {
- super(name, numParallelSubtasks, subtaskIndex, userCodeClassLoader, executionConfig, cpTasks);
+ public DistributedRuntimeUDFContext(String name, int numParallelSubtasks, int subtaskIndex, ClassLoader userCodeClassLoader,
+ ExecutionConfig executionConfig, Map<String, Future<Path>> cpTasks, Map<String, Accumulator<?,?>> accumulators) {
+ super(name, numParallelSubtasks, subtaskIndex, userCodeClassLoader, executionConfig, accumulators, cpTasks);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/RuntimeEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/RuntimeEnvironment.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/RuntimeEnvironment.java
index 5ab0150..f166c36 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/RuntimeEnvironment.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/RuntimeEnvironment.java
@@ -21,10 +21,9 @@ package org.apache.flink.runtime.taskmanager;
import akka.actor.ActorRef;
import org.apache.flink.api.common.JobID;
-import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
-import org.apache.flink.runtime.accumulators.AccumulatorEvent;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.broadcast.BroadcastVariableManager;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
@@ -34,12 +33,10 @@ import org.apache.flink.runtime.io.network.partition.consumer.InputGate;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.tasks.InputSplitProvider;
import org.apache.flink.runtime.memorymanager.MemoryManager;
-import org.apache.flink.runtime.messages.accumulators.ReportAccumulatorResult;
import org.apache.flink.runtime.messages.checkpoint.AcknowledgeCheckpoint;
import org.apache.flink.runtime.state.StateHandle;
import org.apache.flink.runtime.util.SerializedValue;
-import java.io.IOException;
import java.util.Map;
import java.util.concurrent.Future;
@@ -76,7 +73,9 @@ public class RuntimeEnvironment implements Environment {
private final InputGate[] inputGates;
private final ActorRef jobManagerActor;
-
+
+ private final AccumulatorRegistry accumulatorRegistry;
+
// ------------------------------------------------------------------------
public RuntimeEnvironment(JobID jobId, JobVertexID jobVertexId, ExecutionAttemptID executionId,
@@ -86,6 +85,7 @@ public class RuntimeEnvironment implements Environment {
ClassLoader userCodeClassLoader,
MemoryManager memManager, IOManager ioManager,
BroadcastVariableManager bcVarManager,
+ AccumulatorRegistry accumulatorRegistry,
InputSplitProvider splitProvider,
Map<String, Future<Path>> distCacheEntries,
ResultPartitionWriter[] writers,
@@ -93,7 +93,7 @@ public class RuntimeEnvironment implements Environment {
ActorRef jobManagerActor) {
checkArgument(parallelism > 0 && subtaskIndex >= 0 && subtaskIndex < parallelism);
-
+
this.jobId = checkNotNull(jobId);
this.jobVertexId = checkNotNull(jobVertexId);
this.executionId = checkNotNull(executionId);
@@ -107,6 +107,7 @@ public class RuntimeEnvironment implements Environment {
this.memManager = checkNotNull(memManager);
this.ioManager = checkNotNull(ioManager);
this.bcVarManager = checkNotNull(bcVarManager);
+ this.accumulatorRegistry = checkNotNull(accumulatorRegistry);
this.splitProvider = checkNotNull(splitProvider);
this.distCacheEntries = checkNotNull(distCacheEntries);
this.writers = checkNotNull(writers);
@@ -183,6 +184,11 @@ public class RuntimeEnvironment implements Environment {
}
@Override
+ public AccumulatorRegistry getAccumulatorRegistry() {
+ return accumulatorRegistry;
+ }
+
+ @Override
public InputSplitProvider getInputSplitProvider() {
return splitProvider;
}
@@ -213,20 +219,6 @@ public class RuntimeEnvironment implements Environment {
}
@Override
- public void reportAccumulators(Map<String, Accumulator<?, ?>> accumulators) {
- AccumulatorEvent evt;
- try {
- evt = new AccumulatorEvent(getJobID(), accumulators);
- }
- catch (IOException e) {
- throw new RuntimeException("Cannot serialize accumulators to send them to JobManager", e);
- }
-
- ReportAccumulatorResult accResult = new ReportAccumulatorResult(jobId, executionId, evt);
- jobManagerActor.tell(accResult, ActorRef.noSender());
- }
-
- @Override
public void acknowledgeCheckpoint(long checkpointId) {
acknowledgeCheckpoint(checkpointId, null);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
index 616998c..13a2ace 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
@@ -25,6 +25,7 @@ import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.cache.DistributedCache;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.blob.BlobKey;
import org.apache.flink.runtime.broadcast.BroadcastVariableManager;
import org.apache.flink.runtime.deployment.InputGateDeploymentDescriptor;
@@ -100,10 +101,10 @@ public class Task implements Runnable {
/** The class logger. */
private static final Logger LOG = LoggerFactory.getLogger(Task.class);
-
+
/** The tread group that contains all task threads */
private static final ThreadGroup TASK_THREADS_GROUP = new ThreadGroup("Flink Task Threads");
-
+
/** For atomic state updates */
private static final AtomicReferenceFieldUpdater<Task, ExecutionState> STATE_UPDATER =
AtomicReferenceFieldUpdater.newUpdater(Task.class, ExecutionState.class, "executionState");
@@ -176,13 +177,16 @@ public class Task implements Runnable {
/** The library cache, from which the task can request its required JAR files */
private final LibraryCacheManager libraryCache;
-
+
/** The cache for user-defined files that the invokable requires */
private final FileCache fileCache;
-
+
/** The gateway to the network stack, which handles inputs and produced results */
private final NetworkEnvironment network;
+ /** The registry of this task which enables live reporting of accumulators */
+ private final AccumulatorRegistry accumulatorRegistry;
+
/** The thread that executes the task */
private final Thread executingThread;
@@ -194,10 +198,10 @@ public class Task implements Runnable {
/** atomic flag that makes sure the invokable is canceled exactly once upon error */
private final AtomicBoolean invokableHasBeenCanceled;
-
+
/** The invokable of this task, if initialized */
private volatile AbstractInvokable invokable;
-
+
/** The current execution state of the task */
private volatile ExecutionState executionState = ExecutionState.CREATED;
@@ -245,12 +249,13 @@ public class Task implements Runnable {
this.memoryManager = checkNotNull(memManager);
this.ioManager = checkNotNull(ioManager);
- this.broadcastVariableManager =checkNotNull(bcVarManager);
+ this.broadcastVariableManager = checkNotNull(bcVarManager);
+ this.accumulatorRegistry = new AccumulatorRegistry(jobId, executionId);
this.jobManager = checkNotNull(jobManagerActor);
this.taskManager = checkNotNull(taskManagerActor);
this.actorAskTimeout = new Timeout(checkNotNull(actorAskTimeout));
-
+
this.libraryCache = checkNotNull(libraryCache);
this.fileCache = checkNotNull(fileCache);
this.network = checkNotNull(networkEnvironment);
@@ -361,6 +366,10 @@ public class Task implements Runnable {
return inputGatesById.get(id);
}
+ public AccumulatorRegistry getAccumulatorRegistry() {
+ return accumulatorRegistry;
+ }
+
public Thread getExecutingThread() {
return executingThread;
}
@@ -499,7 +508,8 @@ public class Task implements Runnable {
Environment env = new RuntimeEnvironment(jobId, vertexId, executionId,
taskName, taskNameWithSubtask, subtaskIndex, parallelism,
jobConfiguration, taskConfiguration,
- userCodeClassLoader, memoryManager, ioManager, broadcastVariableManager,
+ userCodeClassLoader, memoryManager, ioManager,
+ broadcastVariableManager, accumulatorRegistry,
splitProvider, distributedCacheEntries,
writers, inputGates, jobManager);
@@ -518,7 +528,7 @@ public class Task implements Runnable {
// get our private reference onto the stack (be safe against concurrent changes)
SerializedValue<StateHandle<?>> operatorState = this.operatorState;
-
+
if (operatorState != null) {
if (invokable instanceof OperatorStateCarrier) {
try {
@@ -553,7 +563,7 @@ public class Task implements Runnable {
if (!STATE_UPDATER.compareAndSet(this, ExecutionState.DEPLOYING, ExecutionState.RUNNING)) {
throw new CancelTaskException();
}
-
+
// notify everyone that we switched to running. especially the TaskManager needs
// to know this!
notifyObservers(ExecutionState.RUNNING, null);
@@ -653,7 +663,7 @@ public class Task implements Runnable {
finally {
try {
LOG.info("Freeing task resources for " + taskNameWithSubtask);
-
+
// stop the async dispatcher.
// copy dispatcher reference to stack, against concurrent release
ExecutorService dispatcher = this.asyncCallDispatcher;
@@ -867,15 +877,15 @@ public class Task implements Runnable {
*/
public void triggerCheckpointBarrier(final long checkpointID, final long checkpointTimestamp) {
AbstractInvokable invokable = this.invokable;
-
+
if (executionState == ExecutionState.RUNNING && invokable != null) {
if (invokable instanceof CheckpointedOperator) {
-
+
// build a local closure
final CheckpointedOperator checkpointer = (CheckpointedOperator) invokable;
final Logger logger = LOG;
final String taskName = taskNameWithSubtask;
-
+
Runnable runnable = new Runnable() {
@Override
public void run() {
@@ -1038,7 +1048,7 @@ public class Task implements Runnable {
public String toString() {
return getTaskNameWithSubtasks() + " [" + executionState + ']';
}
-
+
// ------------------------------------------------------------------------
// Task Names
// ------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/TaskExecutionState.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/TaskExecutionState.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/TaskExecutionState.java
index 6c85ab5..0637017 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/TaskExecutionState.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/TaskExecutionState.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.taskmanager;
import java.util.Arrays;
+import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.api.common.JobID;
@@ -52,8 +53,11 @@ public class TaskExecutionState implements java.io.Serializable {
// class may not be part of the system class loader.
private transient Throwable cachedError;
+ /** Serialized flink and user-defined accumulators */
+ private final AccumulatorSnapshot accumulators;
+
/**
- * Creates a new task execution state update, with no attached exception.
+ * Creates a new task execution state update, with no attached exception and no accumulators.
*
* @param jobID
* the ID of the job the task belongs to
@@ -63,13 +67,28 @@ public class TaskExecutionState implements java.io.Serializable {
* the execution state to be reported
*/
public TaskExecutionState(JobID jobID, ExecutionAttemptID executionId, ExecutionState executionState) {
- this(jobID, executionId, executionState, null);
+ this(jobID, executionId, executionState, null, null);
}
-
+
+ /**
+ * Creates a new task execution state update, with an attached exception but no accumulators.
+ *
+ * @param jobID
+ * the ID of the job the task belongs to
+ * @param executionId
+ * the ID of the task execution whose state is to be reported
+ * @param executionState
+ * the execution state to be reported
+ */
+ public TaskExecutionState(JobID jobID, ExecutionAttemptID executionId,
+ ExecutionState executionState, Throwable error) {
+ this(jobID, executionId, executionState, error, null);
+ }
+
/**
* Creates a new task execution state update, with an attached exception.
* This constructor may never throw an exception.
- *
+ *
* @param jobID
* the ID of the job the task belongs to
* @param executionId
@@ -78,11 +97,15 @@ public class TaskExecutionState implements java.io.Serializable {
* the execution state to be reported
* @param error
* an optional error
+ * @param accumulators
+ * The flink and user-defined accumulators which may be null.
*/
public TaskExecutionState(JobID jobID, ExecutionAttemptID executionId,
- ExecutionState executionState, Throwable error) {
+ ExecutionState executionState, Throwable error,
+ AccumulatorSnapshot accumulators) {
- if (jobID == null || executionId == null || executionState == null) {
+
+ if (jobID == null || executionId == null || executionState == null) {
throw new NullPointerException();
}
@@ -90,6 +113,7 @@ public class TaskExecutionState implements java.io.Serializable {
this.executionId = executionId;
this.executionState = executionState;
this.cachedError = error;
+ this.accumulators = accumulators;
if (error != null) {
byte[] serializedError;
@@ -178,6 +202,13 @@ public class TaskExecutionState implements java.io.Serializable {
return this.jobID;
}
+ /**
+ * Gets flink and user-defined accumulators in serialized form.
+ */
+ public AccumulatorSnapshot getAccumulators() {
+ return accumulators;
+ }
+
// --------------------------------------------------------------------------------------------
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/java/org/apache/flink/runtime/util/SerializedValue.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/SerializedValue.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/SerializedValue.java
index f5e897b..6a5468a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/util/SerializedValue.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/SerializedValue.java
@@ -29,7 +29,7 @@ import java.util.Arrays;
* special class loader, the deserialization fails with a {@code ClassNotFoundException}.
*
* To work around that issue, the SerializedValue serialized data immediately into a byte array.
- * When send through RPC or another service that uses serialization, the only the byte array is
+ * When send through RPC or another service that uses serialization, only the byte array is
* transferred. The object is deserialized later (upon access) and requires the accessor to
* provide the corresponding class loader.
*
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
index 3b4ce15..8823041 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/jobmanager/JobManager.scala
@@ -29,39 +29,39 @@ import org.apache.flink.api.common.{ExecutionConfig, JobID}
import org.apache.flink.configuration.{ConfigConstants, Configuration, GlobalConfiguration}
import org.apache.flink.core.io.InputSplitAssigner
import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult
-import org.apache.flink.runtime.akka.AkkaUtils
import org.apache.flink.runtime.blob.BlobServer
import org.apache.flink.runtime.client._
-import org.apache.flink.runtime.execution.librarycache.BlobLibraryCacheManager
import org.apache.flink.runtime.executiongraph.{ExecutionGraph, ExecutionJobVertex}
-import org.apache.flink.runtime.instance.InstanceManager
-import org.apache.flink.runtime.jobgraph.{JobGraph, JobStatus, JobVertexID}
-import org.apache.flink.runtime.jobmanager.accumulators.AccumulatorManager
-import org.apache.flink.runtime.jobmanager.scheduler.{Scheduler => FlinkScheduler}
import org.apache.flink.runtime.jobmanager.web.WebInfoServer
import org.apache.flink.runtime.messages.ArchiveMessages.ArchiveExecutionGraph
import org.apache.flink.runtime.messages.ExecutionGraphMessages.JobStatusChanged
-import org.apache.flink.runtime.messages.JobManagerMessages._
import org.apache.flink.runtime.messages.Messages.{Acknowledge, Disconnect}
-import org.apache.flink.runtime.messages.RegistrationMessages._
-import org.apache.flink.runtime.messages.TaskManagerMessages.{Heartbeat, SendStackTrace}
import org.apache.flink.runtime.messages.TaskMessages.{PartitionState, UpdateTaskExecutionState}
import org.apache.flink.runtime.messages.accumulators._
import org.apache.flink.runtime.messages.checkpoint.{AbstractCheckpointMessage, AcknowledgeCheckpoint}
-import org.apache.flink.runtime.net.NetUtils
import org.apache.flink.runtime.process.ProcessReaper
import org.apache.flink.runtime.security.SecurityUtils
import org.apache.flink.runtime.security.SecurityUtils.FlinkSecuredRunner
import org.apache.flink.runtime.taskmanager.TaskManager
-import org.apache.flink.runtime.util.{EnvironmentInformation, SerializedValue, ZooKeeperUtil}
-import org.apache.flink.runtime.{ActorLogMessages, ActorSynchronousLogging, StreamingMode}
+import org.apache.flink.runtime.util.ZooKeeperUtil
+import org.apache.flink.runtime.util.{SerializedValue, EnvironmentInformation}
+import org.apache.flink.runtime.{StreamingMode, ActorSynchronousLogging, ActorLogMessages}
+import org.apache.flink.runtime.akka.AkkaUtils
+import org.apache.flink.runtime.execution.librarycache.BlobLibraryCacheManager
+import org.apache.flink.runtime.instance.InstanceManager
+import org.apache.flink.runtime.jobgraph.{JobVertexID, JobGraph, JobStatus}
+import org.apache.flink.runtime.jobmanager.scheduler.{Scheduler => FlinkScheduler}
+import org.apache.flink.runtime.messages.JobManagerMessages._
+import org.apache.flink.runtime.messages.RegistrationMessages._
+import org.apache.flink.runtime.messages.TaskManagerMessages.{SendStackTrace, Heartbeat}
import org.apache.flink.util.{ExceptionUtils, InstantiationUtil}
-import scala.collection.JavaConverters._
import scala.concurrent._
import scala.concurrent.duration._
import scala.concurrent.forkjoin.ForkJoinPool
import scala.language.postfixOps
+import scala.collection.JavaConverters._
+import scala.collection.JavaConversions._
/**
* The job manager is responsible for receiving Flink jobs, scheduling the tasks, gathering the
@@ -97,7 +97,6 @@ class JobManager(
protected val scheduler: FlinkScheduler,
protected val libraryCacheManager: BlobLibraryCacheManager,
protected val archive: ActorRef,
- protected val accumulatorManager: AccumulatorManager,
protected val defaultExecutionRetries: Int,
protected val delayBetweenRetries: Long,
protected val timeout: FiniteDuration,
@@ -221,7 +220,6 @@ class JobManager(
originalSender ! result
}(context.dispatcher)
- sender ! true
case None => log.error("Cannot find execution graph for ID " +
s"${taskExecutionState.getJobID} to change state to " +
s"${taskExecutionState.getExecutionState}.")
@@ -298,7 +296,7 @@ class JobManager(
newJobStatus match {
case JobStatus.FINISHED =>
val accumulatorResults: java.util.Map[String, SerializedValue[AnyRef]] = try {
- accumulatorManager.getJobAccumulatorResultsSerialized(jobID)
+ executionGraph.getAccumulatorsSerialized
} catch {
case e: Exception =>
log.error(s"Cannot fetch serialized accumulators for job $jobID", e)
@@ -402,13 +400,22 @@ class JobManager(
import scala.collection.JavaConverters._
sender ! RegisteredTaskManagers(instanceManager.getAllRegisteredInstances.asScala)
- case Heartbeat(instanceID, metricsReport) =>
- try {
- log.debug(s"Received hearbeat message from $instanceID.")
- instanceManager.reportHeartBeat(instanceID, metricsReport)
- } catch {
- case t: Throwable => log.error(s"Could not report heart beat from ${sender().path}.", t)
- }
+ case Heartbeat(instanceID, metricsReport, accumulators) =>
+ log.debug(s"Received hearbeat message from $instanceID.")
+
+ Future {
+ accumulators foreach {
+ case accumulators =>
+ currentJobs.get(accumulators.getJobID) match {
+ case Some((jobGraph, jobInfo)) =>
+ jobGraph.updateAccumulators(accumulators)
+ case None =>
+ // ignore accumulator values for old job
+ }
+ }
+ }(context.dispatcher)
+
+ instanceManager.reportHeartBeat(instanceID, metricsReport)
case message: AccumulatorMessage => handleAccumulatorMessage(message)
@@ -676,33 +683,18 @@ class JobManager(
* @param message The accumulator message.
*/
private def handleAccumulatorMessage(message: AccumulatorMessage): Unit = {
-
message match {
- case ReportAccumulatorResult(jobId, _, accumulatorEvent) =>
- val classLoader = try {
- libraryCacheManager.getClassLoader(jobId)
- } catch {
- case e: Exception =>
- log.error("Dropping accumulators. No class loader available for job " + jobId, e)
- return
- }
-
- if (classLoader != null) {
- try {
- val accumulators = accumulatorEvent.deserializeValue(classLoader)
- accumulatorManager.processIncomingAccumulators(jobId, accumulators)
- }
- catch {
- case e: Exception => log.error("Cannot update accumulators for job " + jobId, e)
- }
- } else {
- log.error("Dropping accumulators. No class loader available for job " + jobId)
- }
case RequestAccumulatorResults(jobID) =>
try {
- val accumulatorValues: java.util.Map[String, SerializedValue[Object]] =
- accumulatorManager.getJobAccumulatorResultsSerialized(jobID)
+ val accumulatorValues: java.util.Map[String, SerializedValue[Object]] = {
+ currentJobs.get(jobID) match {
+ case Some((graph, jobInfo)) =>
+ graph.getAccumulatorsSerialized
+ case None =>
+ null // TODO check also archive
+ }
+ }
sender() ! AccumulatorResultsFound(jobID, accumulatorValues)
}
@@ -714,8 +706,31 @@ class JobManager(
case RequestAccumulatorResultsStringified(jobId) =>
try {
- val accumulatorValues: Array[StringifiedAccumulatorResult] =
- accumulatorManager.getJobAccumulatorResultsStringified(jobId)
+ val accumulatorValues: Array[StringifiedAccumulatorResult] = {
+ currentJobs.get(jobId) match {
+ case Some((graph, jobInfo)) =>
+ val accumulators = graph.aggregateUserAccumulators()
+
+ val result: Array[StringifiedAccumulatorResult] = new
+ Array[StringifiedAccumulatorResult](accumulators.size)
+
+ var i = 0
+ accumulators foreach {
+ case (name, accumulator) =>
+ val (typeString, valueString) =
+ if (accumulator != null) {
+ (accumulator.getClass.getSimpleName, accumulator.toString)
+ } else {
+ (null, null)
+ }
+ result(i) = new StringifiedAccumulatorResult(name, typeString, valueString)
+ i += 1
+ }
+ result
+ case None =>
+ null // TODO check also archive
+ }
+ }
sender() ! AccumulatorResultStringsFound(jobId, accumulatorValues)
}
@@ -1058,7 +1073,7 @@ object JobManager {
*/
def createJobManagerComponents(configuration: Configuration)
: (ExecutionContext, InstanceManager, FlinkScheduler, BlobLibraryCacheManager,
- Props, AccumulatorManager, Int, Long, FiniteDuration, Int) = {
+ Props, Int, Long, FiniteDuration, Int) = {
val timeout: FiniteDuration = AkkaUtils.getTimeout(configuration)
@@ -1091,8 +1106,6 @@ object JobManager {
val archiveProps: Props = Props(classOf[MemoryArchivist], archiveCount)
- val accumulatorManager: AccumulatorManager = new AccumulatorManager(Math.min(1, archiveCount))
-
val executionContext = ExecutionContext.fromExecutor(new ForkJoinPool())
var blobServer: BlobServer = null
@@ -1131,7 +1144,6 @@ object JobManager {
scheduler,
libraryCacheManager,
archiveProps,
- accumulatorManager,
executionRetries,
delayBetweenRetries,
timeout,
@@ -1179,7 +1191,6 @@ object JobManager {
scheduler,
libraryCacheManager,
archiveProps,
- accumulatorManager,
executionRetries,
delayBetweenRetries,
timeout,
@@ -1199,7 +1210,6 @@ object JobManager {
scheduler,
libraryCacheManager,
archiver,
- accumulatorManager,
executionRetries,
delayBetweenRetries,
timeout,
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskManagerMessages.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskManagerMessages.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskManagerMessages.scala
index b12f1b5..6cb571c 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskManagerMessages.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/TaskManagerMessages.scala
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.messages
+import org.apache.flink.runtime.accumulators.AccumulatorSnapshot
import org.apache.flink.runtime.instance.InstanceID
/**
@@ -52,8 +53,10 @@ object TaskManagerMessages {
*
* @param instanceID The instance ID of the reporting TaskManager.
* @param metricsReport utf-8 encoded JSON metrics report from the metricRegistry.
+ * @param accumulators Accumulators of tasks serialized as Tuple2[internal, user-defined]
*/
- case class Heartbeat(instanceID: InstanceID, metricsReport: Array[Byte])
+ case class Heartbeat(instanceID: InstanceID, metricsReport: Array[Byte],
+ accumulators: Seq[AccumulatorSnapshot])
// --------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/accumulators/AccumulatorMessages.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/accumulators/AccumulatorMessages.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/accumulators/AccumulatorMessages.scala
index 82c4ab6..015c96e 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/accumulators/AccumulatorMessages.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/messages/accumulators/AccumulatorMessages.scala
@@ -19,8 +19,7 @@
package org.apache.flink.runtime.messages.accumulators
import org.apache.flink.api.common.JobID
-import org.apache.flink.runtime.accumulators.{StringifiedAccumulatorResult, AccumulatorEvent}
-import org.apache.flink.runtime.executiongraph.ExecutionAttemptID
+import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult
import org.apache.flink.runtime.util.SerializedValue
/**
@@ -38,18 +37,6 @@ sealed trait AccumulatorMessage {
sealed trait AccumulatorResultsResponse extends AccumulatorMessage
/**
- * Reports the accumulator results of the individual tasks to the job manager.
- *
- * @param jobID The ID of the job the accumulator belongs to
- * @param executionId The ID of the task execution that the accumulator belongs to.
- * @param accumulatorEvent The serialized accumulators
- */
-case class ReportAccumulatorResult(jobID: JobID,
- executionId: ExecutionAttemptID,
- accumulatorEvent: AccumulatorEvent)
- extends AccumulatorMessage
-
-/**
* Requests the accumulator results of the job identified by [[jobID]] from the job manager.
* The result is sent back to the sender as a [[AccumulatorResultsResponse]] message.
*
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
index 1a35d01..f07fa0c 100644
--- a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
+++ b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
@@ -37,6 +37,7 @@ import grizzled.slf4j.Logger
import org.apache.flink.configuration.{Configuration, ConfigConstants, GlobalConfiguration, IllegalConfigurationException}
import org.apache.flink.runtime.messages.checkpoint.{NotifyCheckpointComplete, TriggerCheckpoint, AbstractCheckpointMessage}
+import org.apache.flink.runtime.accumulators.{AccumulatorSnapshot, AccumulatorRegistry}
import org.apache.flink.runtime.{StreamingMode, ActorSynchronousLogging, ActorLogMessages}
import org.apache.flink.runtime.akka.AkkaUtils
import org.apache.flink.runtime.blob.{BlobService, BlobCache}
@@ -68,6 +69,7 @@ import scala.concurrent.duration._
import scala.concurrent.forkjoin.ForkJoinPool
import scala.util.{Failure, Success}
import scala.collection.JavaConverters._
+import scala.collection.JavaConversions._
import scala.language.postfixOps
@@ -328,7 +330,7 @@ extends Actor with ActorLogMessages with ActorSynchronousLogging {
network.getPartitionManager.releasePartitionsProducedBy(executionID)
} catch {
case t: Throwable => killTaskManagerFatal(
- "Fatal leak: Unable to release intermediate result partition data", t)
+ "Fatal leak: Unable to release intermediate result partition data", t)
}
}
@@ -389,7 +391,7 @@ extends Actor with ActorLogMessages with ActorSynchronousLogging {
} else {
log.debug(s"Cannot find task to cancel for execution ${executionID})")
sender ! new TaskOperationResult(executionID, false,
- "No task with that execution ID was found.")
+ "No task with that execution ID was found.")
}
case PartitionState(taskExecutionId, taskResultId, partitionId, state) =>
@@ -400,7 +402,7 @@ extends Actor with ActorLogMessages with ActorSynchronousLogging {
log.debug(s"Cannot find task $taskExecutionId to respond with partition state.")
}
}
- }
+ }
}
/**
@@ -793,12 +795,12 @@ extends Actor with ActorLogMessages with ActorSynchronousLogging {
// create the task. this does not grab any TaskManager resources or download
// and libraries - the operation does not block
- val execId = tdd.getExecutionId
val task = new Task(tdd, memoryManager, ioManager, network, bcVarManager,
self, jobManagerActor, config.timeout, libCache, fileCache)
log.info(s"Received task ${task.getTaskNameWithSubtasks}")
-
+
+ val execId = tdd.getExecutionId
// add the task to the map
val prevTask = runningTasks.put(execId, task)
if (prevTask != null) {
@@ -898,22 +900,28 @@ extends Actor with ActorLogMessages with ActorSynchronousLogging {
val task = runningTasks.remove(executionID)
if (task != null) {
-
- // the task must be in a terminal state
- if (!task.getExecutionState.isTerminal) {
- try {
- task.failExternally(new Exception("Task is being removed from TaskManager"))
- } catch {
- case e: Exception => log.error("Could not properly fail task", e)
- }
+
+ // the task must be in a terminal state
+ if (!task.getExecutionState.isTerminal) {
+ try {
+ task.failExternally(new Exception("Task is being removed from TaskManager"))
+ } catch {
+ case e: Exception => log.error("Could not properly fail task", e)
}
+ }
+
+ log.info(s"Unregistering task and sending final execution state " +
+ s"${task.getExecutionState} to JobManager for task ${task.getTaskName} " +
+ s"(${task.getExecutionId})")
- log.info(s"Unregistering task and sending final execution state " +
- s"${task.getExecutionState} to JobManager for task ${task.getTaskName} " +
- s"(${task.getExecutionId})")
+ val accumulators = {
+ val registry = task.getAccumulatorRegistry
+ registry.getSnapshot
+ }
- self ! UpdateTaskExecutionState(new TaskExecutionState(
- task.getJobID, task.getExecutionId, task.getExecutionState, task.getFailureCause))
+ self ! UpdateTaskExecutionState(new TaskExecutionState(
+ task.getJobID, task.getExecutionId, task.getExecutionState, task.getFailureCause,
+ accumulators))
}
else {
log.error(s"Cannot find task with ID $executionID to unregister.")
@@ -931,9 +939,20 @@ extends Actor with ActorLogMessages with ActorSynchronousLogging {
private def sendHeartbeatToJobManager(): Unit = {
try {
log.debug("Sending heartbeat to JobManager")
- val report: Array[Byte] = metricRegistryMapper.writeValueAsBytes(metricRegistry)
- currentJobManager foreach {
- jm => jm ! Heartbeat(instanceID, report)
+ 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 ! Heartbeat(instanceID, metricsReport, accumulatorEvents)
}
}
catch {
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/reader/AbstractReaderTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/reader/AbstractReaderTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/reader/AbstractReaderTest.java
index 4e5fb40..14bf022 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/reader/AbstractReaderTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/reader/AbstractReaderTest.java
@@ -18,8 +18,10 @@
package org.apache.flink.runtime.io.network.api.reader;
+import org.apache.flink.api.common.accumulators.LongCounter;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.event.task.TaskEvent;
import org.apache.flink.runtime.io.network.api.EndOfPartitionEvent;
import org.apache.flink.runtime.io.network.api.EndOfSuperstepEvent;
@@ -183,5 +185,10 @@ public class AbstractReaderTest {
protected MockReader(InputGate inputGate) {
super(inputGate);
}
+
+ @Override
+ public void setReporter(AccumulatorRegistry.Reporter reporter) {
+
+ }
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DriverTestBase.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DriverTestBase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DriverTestBase.java
index 9b9609b..0aab5fe 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DriverTestBase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/DriverTestBase.java
@@ -61,7 +61,7 @@ public class DriverTestBase<S extends Function> implements PactTaskContext<S, Re
private final IOManager ioManager;
private final MemoryManager memManager;
-
+
private final List<MutableObjectIterator<Record>> inputs;
private final List<TypeComparator<Record>> comparators;
@@ -105,7 +105,7 @@ public class DriverTestBase<S extends Function> implements PactTaskContext<S, Re
this.perSortFractionMem = (double)perSortMemory/totalMem;
this.ioManager = new IOManagerAsync();
this.memManager = totalMem > 0 ? new DefaultMemoryManager(totalMem,1) : null;
-
+
this.inputs = new ArrayList<MutableObjectIterator<Record>>();
this.comparators = new ArrayList<TypeComparator<Record>>();
this.sorters = new ArrayList<UnilateralSortMerger<Record>>();
@@ -295,7 +295,7 @@ public class DriverTestBase<S extends Function> implements PactTaskContext<S, Re
public IOManager getIOManager() {
return this.ioManager;
}
-
+
@Override
public MemoryManager getMemoryManager() {
return this.memManager;
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java
index 0f62b27..b9cb416 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/testutils/MockEnvironment.java
@@ -22,6 +22,7 @@ import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.broadcast.BroadcastVariableManager;
import org.apache.flink.runtime.execution.Environment;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
@@ -79,6 +80,8 @@ public class MockEnvironment implements Environment {
private final BroadcastVariableManager bcVarManager = new BroadcastVariableManager();
+ private final AccumulatorRegistry accumulatorRegistry;
+
private final int bufferSize;
public MockEnvironment(long memorySize, MockInputSplitProvider inputSplitProvider, int bufferSize) {
@@ -91,6 +94,8 @@ public class MockEnvironment implements Environment {
this.ioManager = new IOManagerAsync();
this.inputSplitProvider = inputSplitProvider;
this.bufferSize = bufferSize;
+
+ this.accumulatorRegistry = new AccumulatorRegistry(jobID, getExecutionId());
}
public IteratorWrappingTestSingleInputGate<Record> addInput(MutableObjectIterator<Record> inputIterator) {
@@ -259,8 +264,8 @@ public class MockEnvironment implements Environment {
}
@Override
- public void reportAccumulators(Map<String, Accumulator<?, ?>> accumulators) {
- // discard, this is only for testing
+ public AccumulatorRegistry getAccumulatorRegistry() {
+ return this.accumulatorRegistry;
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
index e9e761c..bd36dd4 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskTest.java
@@ -25,6 +25,7 @@ import akka.actor.Props;
import com.google.common.collect.Maps;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.blob.BlobKey;
import org.apache.flink.runtime.broadcast.BroadcastVariableManager;
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
index 219e5ae..f2535fa 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
@@ -65,7 +65,6 @@ class TestingCluster(userConfiguration: Configuration,
scheduler,
libraryCacheManager,
_,
- accumulatorManager,
executionRetries,
delayBetweenRetries,
timeout,
@@ -82,7 +81,6 @@ class TestingCluster(userConfiguration: Configuration,
scheduler,
libraryCacheManager,
archive,
- accumulatorManager,
executionRetries,
delayBetweenRetries,
timeout,
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
index 5747b7e..6d316ca 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
@@ -147,6 +147,16 @@ trait TestingJobManager extends ActorLogMessages with WrapAsScala {
case None => sender ! WorkingTaskManager(None)
}
+ case RequestAccumulatorValues(jobID) =>
+
+ val (flinkAccumulators, userAccumulators) = currentJobs.get(jobID) match {
+ case Some((graph, jobInfo)) =>
+ (graph.getFlinkAccumulators, graph.aggregateUserAccumulators)
+ case None => null
+ }
+
+ sender ! RequestAccumulatorValuesResponse(jobID, flinkAccumulators, userAccumulators)
+
case NotifyWhenJobStatus(jobID, state) =>
val jobStatusListener = waitForJobStatus.getOrElseUpdate(jobID,
scala.collection.mutable.HashMap[JobStatus, Set[ActorRef]]())
http://git-wip-us.apache.org/repos/asf/flink/blob/8261ed54/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerMessages.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerMessages.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerMessages.scala
index 241c6c0..46e8486 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerMessages.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManagerMessages.scala
@@ -20,9 +20,12 @@ package org.apache.flink.runtime.testingUtils
import akka.actor.ActorRef
import org.apache.flink.api.common.JobID
-import org.apache.flink.runtime.executiongraph.ExecutionGraph
+import org.apache.flink.runtime.accumulators.AccumulatorRegistry
+import org.apache.flink.runtime.executiongraph.{ExecutionAttemptID, ExecutionGraph}
import org.apache.flink.runtime.instance.InstanceGateway
import org.apache.flink.runtime.jobgraph.JobStatus
+import java.util.Map
+import org.apache.flink.api.common.accumulators.Accumulator
object TestingJobManagerMessages {
@@ -53,4 +56,9 @@ object TestingJobManagerMessages {
case class NotifyWhenTaskManagerTerminated(taskManager: ActorRef)
case class TaskManagerTerminated(taskManager: ActorRef)
+
+ case class RequestAccumulatorValues(jobID: JobID)
+ case class RequestAccumulatorValuesResponse(jobID: JobID,
+ flinkAccumulators: Map[ExecutionAttemptID, Map[AccumulatorRegistry.Metric, Accumulator[_,_]]],
+ userAccumulators: Map[String, Accumulator[_,_]])
}