[
https://issues.apache.org/jira/browse/FLINK-14740?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
lining updated FLINK-14740:
---------------------------
Description:
Now OperatorMetricGroup which in batch job use VertexId as OperatorId. For
chain operator, they'll use the same id, if two chain operators which have same
name. I We could update in JobGraphGenerator.compileJobGraph
{code:java}
for (TaskInChain tic : this.chainedTasksInSequence) {
TaskConfig t = new TaskConfig(tic.getContainingVertex().getConfiguration());
Integer nodeId = tic.getPlanNode().getOptimizerNode().getId();
OperatorID operatorID = this.nodeId2OperatorId.get(nodeId);
if(operatorID == null) {
operatorID = new OperatorID();
this.nodeId2OperatorId.put(nodeId, operatorID);
}
t.addChainedTask(tic.getChainedTask(), tic.getTaskConfig(),
tic.getTaskName(), operatorID.toString());
}
{code}
Then we could get id from TaskInfo.
was:
Now OperatorMetricGroup which in batch job use VertexId as OperatorId. For
chain operator, they'll use the same id, if two chain operators which have same
name. It's no good. We could update in JobGraphGenerator.compileJobGraph
{code:java}
for (TaskInChain tic : this.chainedTasksInSequence) {
TaskConfig t = new TaskConfig(tic.getContainingVertex().getConfiguration());
Integer nodeId = tic.getPlanNode().getOptimizerNode().getId();
OperatorID operatorID = this.nodeId2OperatorId.get(nodeId);
if(operatorID == null) {
operatorID = new OperatorID();
this.nodeId2OperatorId.put(nodeId, operatorID);
}
t.addChainedTask(tic.getChainedTask(), tic.getTaskConfig(),
tic.getTaskName(), operatorID.toString());
}
{code}
Then we could get id from TaskInfo.
> Create OperatorID for OperatorMetricGroup which in batch job
> -------------------------------------------------------------
>
> Key: FLINK-14740
> URL: https://issues.apache.org/jira/browse/FLINK-14740
> Project: Flink
> Issue Type: Wish
> Components: Runtime / Metrics
> Reporter: lining
> Priority: Major
>
> Now OperatorMetricGroup which in batch job use VertexId as OperatorId. For
> chain operator, they'll use the same id, if two chain operators which have
> same name. I We could update in JobGraphGenerator.compileJobGraph
> {code:java}
> for (TaskInChain tic : this.chainedTasksInSequence) {
> TaskConfig t = new
> TaskConfig(tic.getContainingVertex().getConfiguration());
> Integer nodeId = tic.getPlanNode().getOptimizerNode().getId();
> OperatorID operatorID = this.nodeId2OperatorId.get(nodeId);
> if(operatorID == null) {
> operatorID = new OperatorID();
> this.nodeId2OperatorId.put(nodeId, operatorID);
> }
> t.addChainedTask(tic.getChainedTask(), tic.getTaskConfig(),
> tic.getTaskName(), operatorID.toString());
> }
> {code}
> Then we could get id from TaskInfo.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
