Dolphinscheduler DAG核心源码剖析

背景描述

注意 : 在 Dolphinscheduler 中，离线任务是有完备的声明周期的，比如说停止、暂停、暂停恢复、重跑等等，都是以DAG(有向无环图的形式举行任务组织)T+1离线任务的。
Dolphinscheduler DAG实现

org.apache.dolphinscheduler.common.graph.DAG
DAG三个重要的数据结构 :

1. // 顶点信息
2. private final Map<Node, NodeInfo> nodesMap;
4. // 边关联信息，作用是记录顶点和边的关系，可以找到叶子节点，也可以获取下游节点
5. private final Map<Node, Map<Node, EdgeInfo>> edgesMap;
7. // 反向边关联信息，作用是可以快速找到入度为0的节点(起始节点)，也可以获取上游节点
8. private final Map<Node, Map<Node, EdgeInfo>> reverseEdgesMap;

复制代码

如下示例 :

1. DAG<String, String, String> graph = new DAG<>();
2. graph.addNode("A", "A");
3. graph.addNode("B", "B");
4. graph.addNode("C", "C");
6. // 添加一个B -> C的边，当前A还飘着呢
7. graph.addEdge("B", "C");
9. // 如果添加A -> B，其实就是会从B开始一直到子节点，看有没有可连接的线到A，如果有，说明这个A -> B的边添加不得，因为会形成环，否则就可以添加
10. graph.addEdge("A", "B");

复制代码

源码分析 :
org.apache.dolphinscheduler.common.graph.DAG#addEdge

1. public boolean addEdge(Node fromNode, Node toNode, EdgeInfo edge, boolean createNode) {
2.     lock.writeLock().lock();
4.     try {
5.         // TODO 是否可以添加该边
6.         if (!isLegalAddEdge(fromNode, toNode, createNode)) {
7.             log.error("serious error: add edge({} -> {}) is invalid, cause cycle！", fromNode, toNode);
8.             return false;
9.         }
11.         // TODO 添加节点
12.         addNodeIfAbsent(fromNode, null);
13.         addNodeIfAbsent(toNode, null);
15.         // TODO 添加边
16.         addEdge(fromNode, toNode, edge, edgesMap);
17.         addEdge(toNode, fromNode, edge, reverseEdgesMap);
19.         return true;
20.     } finally {
21.         lock.writeLock().unlock();
22.     }
23. }
26. private boolean isLegalAddEdge(Node fromNode, Node toNode, boolean createNode) {
27.     // TODO 如果fromNode和toNode两个是同一个顶点，这个边是不能添加的
28.     if (fromNode.equals(toNode)) {
29.         log.error("edge fromNode({}) can't equals toNode({})", fromNode, toNode);
30.         return false;
31.     }
33.     // TODO 这里其实就是想说，不是创建节点，也就是说要求fromNode和toNode是需要存在的顶点
34.     if (!createNode) {
35.         if (!containsNode(fromNode) || !containsNode(toNode)) {
36.             log.error("edge fromNode({}) or toNode({}) is not in vertices map", fromNode, toNode);
37.             return false;
38.         }
39.     }
41.     // Whether an edge can be successfully added(fromNode -> toNode),need to determine whether the
42.     // DAG has cycle!
43.     // TODO 这里获取节点的数量
44.     int verticesCount = getNodesCount();
46.     Queue<Node> queue = new LinkedList<>();
48.     // TODO 将toNode放入到queue中
49.     queue.add(toNode);
51.     // if DAG doesn't find fromNode, it's not has cycle!
52.     // TODO 当queue不为空，这里肯定就不为空了
53.     while (!queue.isEmpty() && (--verticesCount > 0)) {
54.         // TODO 获取队列里面的元素
55.         Node key = queue.poll();
57.         for (Node subsequentNode : getSubsequentNodes(key)) {
58.             // TODO 其实这里判断的是比如说A -> B 有连接的DAG图，传入的是节点B，看B节点的边是不是有A，如果有A说明已经有B -> A的关联了，
59.             // TODO 就不能添加了。如果比如说B的下游节点，比如说 A -> B -> C，这样的话，B的下游节点就是C，C是需要放入queue中的
60.             // TODO 核心思想其实就是要找到它要添加的目标节点的连线，是否有目标节点到源节点的连线存在(这样来判断是否存在环)
61.             if (subsequentNode.equals(fromNode)) {
62.                 return false;
63.             }
65.             queue.add(subsequentNode);
66.         }
67.     }
69.     return true;
70. }

复制代码

Dolphinscheduler DagHelper讲授

DAG类是一个底子通用的DAG工具类，而DagHelper是任务界说、任务界说直接的关系组装成DAG的一个业务工具类。
org.apache.dolphinscheduler.server.master.graph.WorkflowGraphFactory#createWorkflowGraph

1. public IWorkflowGraph createWorkflowGraph(ProcessInstance workflowInstance) throws Exception {
3.     // TODO 这里其实就是获取的流程实例对应的任务数和之间的关系
4.     List<ProcessTaskRelation> processTaskRelations = processService.findRelationByCode(
5.             workflowInstance.getProcessDefinitionCode(),
6.             workflowInstance.getProcessDefinitionVersion());
8.     // TODO 获取对应的任务定义log
9.     List<TaskDefinitionLog> taskDefinitionLogs = taskDefinitionLogDao.queryTaskDefineLogList(processTaskRelations);
11.     // TODO 获取TaskNode
12.     List<TaskNode> taskNodeList = processService.transformTask(processTaskRelations, taskDefinitionLogs);
14.     // generate process to get DAG info
15.     // TODO 这里其实解析的是是否自己手动指定的启动节点列表，默认不会
16.     List<Long> recoveryTaskNodeCodeList = getRecoveryTaskNodeCodeList(workflowInstance.getCommandParam());
18.     // TODO 如果 默认startNodeNameList为空
19.     List<Long> startNodeNameList = parseStartNodeName(workflowInstance.getCommandParam());
21.     // TODO 构建ProcessDag对象实例
22.     ProcessDag processDag = DagHelper.generateFlowDag(
23.             taskNodeList,
24.             startNodeNameList,
25.             recoveryTaskNodeCodeList,
26.             workflowInstance.getTaskDependType());
28.     if (processDag == null) {
29.         log.error("ProcessDag is null");
30.         throw new IllegalArgumentException("Create WorkflowGraph failed, ProcessDag is null");
31.     }
33.     // TODO 生成DAG
34.     DAG<Long, TaskNode, TaskNodeRelation> dagGraph = DagHelper.buildDagGraph(processDag);
35.     log.debug("Build dag success, dag: {}", dagGraph);
37.     // TODO 使用WorkflowGraph来封装任务节点列表和dagGraph
38.     return new WorkflowGraph(taskNodeList, dagGraph);
39. }

复制代码

org.apache.dolphinscheduler.service.utils.DagHelper#generateFlowDag

1. public static ProcessDag generateFlowDag(
2.                                              List<TaskNode> totalTaskNodeList,
3.                                              List<Long> startNodeNameList,
4.                                              List<Long> recoveryNodeCodeList,
5.                                              TaskDependType depNodeType) throws Exception {
7.     // TODO 其实就是拿到所有的节点
8.     List<TaskNode> destTaskNodeList =
9.             generateFlowNodeListByStartNode(
10.                     totalTaskNodeList, startNodeNameList, recoveryNodeCodeList, depNodeType);
12.     if (destTaskNodeList.isEmpty()) {
13.         return null;
14.     }
16.     // TODO 获取任务节点之前的关系
17.     List<TaskNodeRelation> taskNodeRelations = generateRelationListByFlowNodes(destTaskNodeList);
19.     // TODO 其实就是实例化一个ProcessDag
20.     ProcessDag processDag = new ProcessDag();
21.     // TODO 设置DAG的边
22.     processDag.setEdges(taskNodeRelations);
23.     // TODO 设置DAG的顶点
24.     processDag.setNodes(destTaskNodeList);
25.     return processDag;
26. }

复制代码

设置了destTaskNodeList和taskNodeRelations
org.apache.dolphinscheduler.service.utils.DagHelper#buildDagGraph

1. public static DAG<Long, TaskNode, TaskNodeRelation> buildDagGraph(ProcessDag processDag) {
3.     DAG<Long, TaskNode, TaskNodeRelation> dag = new DAG<>();
5.     // TODO 添加顶点
6.     if (CollectionUtils.isNotEmpty(processDag.getNodes())) {
7.         for (TaskNode node : processDag.getNodes()) {
8.             dag.addNode(node.getCode(), node);
9.         }
10.     }
12.     // TODO 添加边
13.     if (CollectionUtils.isNotEmpty(processDag.getEdges())) {
14.         for (TaskNodeRelation edge : processDag.getEdges()) {
15.             dag.addEdge(edge.getStartNode(), edge.getEndNode());
16.         }
17.     }
18.     return dag;
19. }

复制代码

转载自 Journey
原文链接：https://segmentfault.com/a/1190000045117764

本文由 白鲸开源 提供发布支持！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。