neo4j实现表字段级血缘关系

需求背景

需要在前端页面展示当前表字段的所有上下游血缘关系，以进一步做数据诊断治理。大致效果图如下：

首先这里解释什么是表字段血缘关系，SQL 示例：

1. CREATE TABLE IF NOT EXISTS table_b
2. AS SELECT order_id, order_status FROM table_a;

复制代码

如上 DDL 语句中，创建的 table_b 的 order_id 和 order_status 字段来源于 table_a，代表table_a 就是 table_b 的来源表，也叫上游表，table_b 就是 table_a 下游表，另外 table_a.order_id 就是 table_b.order_id 的上游字段，它们之间就存在血缘关系。

1. INSERT INTO table_c
2. SELECT a.order_id, b.order_status
3. FROM table_a a JOIN table_b b ON a.order_id = b.order_id;

复制代码

如上 DML 语句中，table_c 的 order_id 字段来源于 table_a，而 order_status 来源于 table_b，表示 table_c 和 table_a、table_b 之间也存在血缘关系。
由上也可看出想要存储血缘关系，还需要先解析 sql，这块儿主要使用了开源项目 calcite 的解析器，这篇文章不再展开，本篇主要讲如何存储和如何展示
环境配置

参考另一篇：springboot 配置内嵌式 neo4j
Node 数据结构定义

因为要展示表的字段之间的血缘关系，所以直接将表字段作为图节点存储，表字段之间的血缘关系就用图节点之间的关系表示，具体 node 定义如下：

1. public class ColumnVertex {
2.   // 唯一键
3.   private String name;
5.   public ColumnVertex(String catalogName, String databaseName, String tableName, String columnName) {
6.     this.name = catalogName + "." + databaseName + "." + tableName + "." + columnName;
7.   }
9.   public String getCatalogName() {
10.     return Long.parseLong(name.split("\\.")[0]);
11.   }
13.   public String getDatabaseName() {
14.     return name.split("\\.")[1];
15.   }
17.   public String getTableName() {
18.     return name.split("\\.")[2];
19.   }
21.   public String getColumnName() {
22.     return name.split("\\.")[3];
23.   }
24. }

复制代码

通用 Service 定义

1. public interface EmbeddedGraphService {
2.     // 添加图节点以及与上游节点之间的关系
3.     void addColumnVertex(ColumnVertex currentVertex, ColumnVertex upstreamVertex);
4.     // 寻找上游节点
5.     List<ColumnVertex> findUpstreamColumnVertex(ColumnVertex currentVertex);
6.     // 寻找下游节点
7.     List<ColumnVertex> findDownstreamColumnVertex(ColumnVertex currentVertex);
8. }

复制代码

Service 实现

1. import javax.annotation.Resource;
2. import org.neo4j.graphdb.GraphDatabaseService;
3. import org.neo4j.graphdb.Result;
4. import org.neo4j.graphdb.Transaction;
5. import org.springframework.stereotype.Service;
7. @Service
8. public class EmbeddedGraphServiceImpl implements EmbeddedGraphService {
10.   @Resource private GraphDatabaseService graphDb;
12.   @Override
13.   public void addColumnVertex(ColumnVertex currentVertex, ColumnVertex upstreamVertex) {
14.     try (Transaction tx = graphDb.beginTx()) {
15.       tx.execute(
16.           "MERGE (c:ColumnVertex {name: $currentName}) MERGE (u:ColumnVertex {name: $upstreamName})"
17.               + " MERGE (u)-[:UPSTREAM]->(c)",
18.           Map.of("currentName", currentVertex.getName(), "upstreamName", upstreamVertex.getName()));
19.       tx.commit();
20.     }
21.   }
23.   @Override
24.   public List<ColumnVertex> findUpstreamColumnVertex(ColumnVertex currentVertex) {
25.     List<ColumnVertex> result = new ArrayList<>();
26.     try (Transaction tx = graphDb.beginTx()) {
27.       Result queryResult =
28.           tx.execute(
29.               "MATCH (u:ColumnVertex)-[:UPSTREAM]->(c:ColumnVertex) WHERE c.name = $name RETURN"
30.                   + " u.name AS name",
31.               Map.of("name", currentVertex.getName()));
32.       while (queryResult.hasNext()) {
33.         Map<String, Object> row = queryResult.next();
34.         result.add(new ColumnVertex().setName((String) row.get("name")));
35.       }
36.       tx.commit();
37.     }
38.     return result;
39.   }
41.   @Override
42.   public List<ColumnVertex> findDownstreamColumnVertex(ColumnVertex currentVertex) {
43.     List<ColumnVertex> result = new ArrayList<>();
44.     try (Transaction tx = graphDb.beginTx()) {
45.       Result queryResult =
46.           tx.execute(
47.               "MATCH (c:ColumnVertex)-[:UPSTREAM]->(d:ColumnVertex) WHERE c.name = $name RETURN"
48.                   + " d.name AS name",
49.               Map.of("name", currentVertex.getName()));
50.       while (queryResult.hasNext()) {
51.         Map<String, Object> row = queryResult.next();
52.         result.add(new ColumnVertex().setName((String) row.get("name")));
53.       }
54.       tx.commit();
55.     }
56.     return result;
57.   }
58. }

复制代码

遍历图节点

实现逻辑：

• restful 接口入参：当前表（catalogName, databaseName, tableName）
  
• 定义返回给前端的数据结构，采用 nodes 和 edges 方式返回，然后前端再根据节点与边关系渲染出完整的血缘关系图
  

1. public class ColumnLineageVO {
2.   List<ColumnLineageNode> nodes;
3.   List<ColumnLineageEdge> edges;
4. }
6. public class ColumnLineageNode {
7.   private String databaseName;
8.   private String tableName;
9.   private List<String> columnNames;
10. }
12. public class ColumnLineageEdge {
13.   private ColumnLineageEdgePoint source;
14.   private ColumnLineageEdgePoint target;
15. }
17. public class ColumnLineageEdgePoint {
18.   private String databaseName;
19.   private String tableName;
20.   private String columnName;
21. }

复制代码

• 查询表字段
  
• 采用递归的方式，利用当前表字段遍历与当前表字段关联的所有上下游图节点
  
• 将所有节点封装成 List ColumnLineageVO 返回给前端
  

1. public ColumnLineageVO getColumnLineage(Table table) {
2.     ColumnLineageVO columnLineageVO = new ColumnLineageVO();
3.     List<ColumnLineageNode> nodes = new ArrayList<>();
4.     List<ColumnLineageEdge> edges = new ArrayList<>();
5.     // Deduplication
6.     Set<String> visitedNodes = new HashSet<>();
7.     Set<String> visitedEdges = new HashSet<>();
8.     Map<String, List<ColumnVertex>> upstreamCache = new HashMap<>();
9.     Map<String, List<ColumnVertex>> downstreamCache = new HashMap<>();
11.     ColumnLineageNode currentNode =
12.         ColumnLineageNode.builder()
13.             .databaseName(table.getDatabaseName())
14.             .tableName(table.getTableName())
15.             .type(TableType.EXTERNAL_TABLE.getDesc())
16.             .build();
17.     nodes.add(currentNode);
18.     visitedNodes.add(currentNode.getDatabaseName() + "." + currentNode.getTableName());
20.     for (String columnName : table.getColumnNames()) {
21.       ColumnVertex currentVertex =
22.           new ColumnVertex(
23.               table.getScriptId(), table.getDatabaseName(), table.getTableName(), columnName);
24.       traverseUpstreamColumnVertex(
25.           currentVertex, nodes, edges, visitedNodes, visitedEdges, upstreamCache);
26.       traverseDownstreamColumnVertex(
27.           currentVertex, nodes, edges, visitedNodes, visitedEdges, downstreamCache);
28.     }
30.     columnLineageVO.setNodes(nodes);
31.     columnLineageVO.setEdges(edges);
32.     return columnLineageVO;
33.   }
35. private void traverseUpstreamColumnVertex(
36.       ColumnVertex currentVertex,
37.       List<ColumnLineageNode> nodes,
38.       List<ColumnLineageEdge> edges,
39.       Set<String> visitedNodes,
40.       Set<String> visitedEdges,
41.       Map<String, List<ColumnVertex>> cache) {
42.     List<ColumnVertex> upstreamVertices;
43.     if (cache.containsKey(currentVertex.getName())) {
44.       upstreamVertices = cache.get(currentVertex.getName());
45.     } else {
46.       upstreamVertices = embeddedGraphService.findUpstreamColumnVertex(currentVertex);
47.       cache.put(currentVertex.getName(), upstreamVertices);
48.     }
49.     for (ColumnVertex upstreamVertex : upstreamVertices) {
50.       String nodeKey = upstreamVertex.getDatabaseName() + "." + upstreamVertex.getTableName();
51.       if (!visitedNodes.contains(nodeKey)) {
52.         ColumnLineageNode upstreamNode =
53.             ColumnLineageNode.builder()
54.                 .databaseName(upstreamVertex.getDatabaseName())
55.                 .tableName(upstreamVertex.getTableName())
56.                 .type(TableType.EXTERNAL_TABLE.getDesc())
57.                 .build();
58.         nodes.add(upstreamNode);
59.         visitedNodes.add(nodeKey);
60.       }
61.       String edgeKey =
62.           upstreamVertex.getDatabaseName()
63.               + upstreamVertex.getTableName()
64.               + upstreamVertex.getColumnName()
65.               + currentVertex.getDatabaseName()
66.               + currentVertex.getTableName()
67.               + currentVertex.getColumnName();
68.       if (!visitedEdges.contains(edgeKey)) {
69.         ColumnLineageEdge edge = createEdge(upstreamVertex, currentVertex);
70.         edges.add(edge);
71.         visitedEdges.add(edgeKey);
72.       }
73.       traverseUpstreamColumnVertex(upstreamVertex, nodes, edges, visitedNodes, visitedEdges, cache);
74.     }
75.   }
76.   
77. private void traverseDownstreamColumnVertex(
78.       ColumnVertex currentVertex,
79.       List<ColumnLineageNode> nodes,
80.       List<ColumnLineageEdge> edges,
81.       Set<String> visitedNodes,
82.       Set<String> visitedEdges,
83.       Map<String, List<ColumnVertex>> cache) {
84.     List<ColumnVertex> downstreamVertices;
85.     if (cache.containsKey(currentVertex.getName())) {
86.       downstreamVertices = cache.get(currentVertex.getName());
87.     } else {
88.       downstreamVertices = embeddedGraphService.findDownstreamColumnVertex(currentVertex);
89.       cache.put(currentVertex.getName(), downstreamVertices);
90.     }
91.     for (ColumnVertex downstreamVertex : downstreamVertices) {
92.       String nodeKey = downstreamVertex.getDatabaseName() + "." + downstreamVertex.getTableName();
93.       if (!visitedNodes.contains(nodeKey)) {
94.         ColumnLineageNode downstreamNode =
95.             ColumnLineageNode.builder()
96.                 .databaseName(downstreamVertex.getDatabaseName())
97.                 .tableName(downstreamVertex.getTableName())
98.                 .type(TableType.EXTERNAL_TABLE.getDesc())
99.                 .build();
100.         nodes.add(downstreamNode);
101.         visitedNodes.add(nodeKey);
102.       }
103.       String edgeKey =
104.           currentVertex.getDatabaseName()
105.               + currentVertex.getTableName()
106.               + currentVertex.getColumnName()
107.               + downstreamVertex.getDatabaseName()
108.               + downstreamVertex.getTableName()
109.               + downstreamVertex.getColumnName();
110.       if (!visitedEdges.contains(edgeKey)) {
111.         ColumnLineageEdge edge = createEdge(currentVertex, downstreamVertex);
112.         edges.add(edge);
113.         visitedEdges.add(edgeKey);
114.       }
115.       traverseDownstreamColumnVertex(
116.           downstreamVertex, nodes, edges, visitedNodes, visitedEdges, cache);
117.     }
118.   }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！