[
https://issues.apache.org/jira/browse/FLINK-2292?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14625232#comment-14625232
]
ASF GitHub Bot commented on FLINK-2292:
---------------------------------------
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/896#discussion_r34504087
--- Diff:
flink-runtime/src/main/java/org/apache/flink/runtime/accumulators/AccumulatorEvent.java
---
@@ -18,32 +18,60 @@
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.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 job. It is
used
+ * 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 AccumulatorEvent extends SerializedValue<Map<String,
Accumulator<?, ?>>> {
+public class AccumulatorEvent implements Serializable {
- private static final long serialVersionUID = 8965894516006882735L;
+ private static final long serialVersionUID = 42L;
- /** JobID for the target job */
private final JobID jobID;
+ private final ExecutionAttemptID executionAttemptID;
+ private final Map<AccumulatorRegistry.Metric, Accumulator<?, ?>>
flinkAccumulators;
+ private final SerializedValue<Map<String, Accumulator<?, ?>>>
userAccumulators;
- public AccumulatorEvent(JobID jobID, Map<String, Accumulator<?, ?>>
accumulators) throws IOException {
- super(accumulators);
+ public AccumulatorEvent(JobID jobID, ExecutionAttemptID
executionAttemptID,
--- End diff --
Regarding the name of this class: I think the name implies a "single
entity" and not a pair of maps etc. Maybe rename it to something along the
lines of AccumulatorSnapshot? getSnapshot is the only method where this is used
anyways, no?
> Report accumulators periodically while job is running
> -----------------------------------------------------
>
> Key: FLINK-2292
> URL: https://issues.apache.org/jira/browse/FLINK-2292
> Project: Flink
> Issue Type: Sub-task
> Components: JobManager, TaskManager
> Reporter: Maximilian Michels
> Assignee: Maximilian Michels
> Fix For: 0.10
>
>
> Accumulators should be sent periodically, as part of the heartbeat that sends
> metrics. This allows them to be updated in real time.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
