有没有办法通过 Neo4j Cypher 查询来跟踪端到端数据沿袭？

Question

我正在使用 Spring-Data 和 SpringBoot 来填充我的 Neo4j 图形数据库。

我定义了以下 Neo4j 实体：

Source 实体 -->

@NodeEntity
public class Source implements Comparable<Source> {

    @GraphId private Long id;

    private String name;
    private SourceType type;
    private String dataStoreName;
    private String dataStoreDesc;

    private Source() {
        // Empty constructor required as of Neo4j API 2.0.5
    };

    public Source(String name, SourceType type, String dataStoreName, String dataStoreDesc) {
        this.name = name;
        this.type = type;
        this.dataStoreName = dataStoreName;
        this.dataStoreDesc = dataStoreDesc;
    }
    @Relationship(type = "CONTAINS", direction = Relationship.UNDIRECTED)
    public Set<Field> fields;

    public void contains(Field field) {
        if (fields == null) {
            fields = new HashSet<Field>();
        }
        fields.add(field);
    }


    /* Getter and Setters */

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public SourceType getType() {
        return type;
    }

    public void setType(SourceType type) {
        this.type = type;
    }

    public String getDataStoreName() {
        return dataStoreName;
    }

    public void setDataStoreName(String dataStoreName) {
        this.dataStoreName = dataStoreName;
    }

    public String getDataStoreDesc() {
        return dataStoreDesc;
    }

    public void setDataStoreDesc(String dataStoreDesc) {
        this.dataStoreDesc = dataStoreDesc;
    }

    public Set<Field> getFields() {
        return fields;
    }

    public void setFields(Set<Field> fields) {
        this.fields = fields;
    }

    @Override
    public int compareTo(Source other) {
        String name = other.getName();
        SourceType type = other.getType();
        if(this.name.equalsIgnoreCase(name) && this.type.equals(type))
            return 0;

        return -1;
    }

    @Override
    public int hashCode() {
        final int prime = 31;
        int result = 1;
        result = prime * result + ((name == null) ? 0 : name.hashCode());
        result = prime * result + ((type == null) ? 0 : type.hashCode());
        return result;
    }

    @Override
    public boolean equals(Object obj) {
        if (this == obj)
            return true;
        if (obj == null)
            return false;
        if (getClass() != obj.getClass())
            return false;
        Source other = (Source) obj;
        if (name == null) {
            if (other.name != null)
                return false;
        } else if (!name.equals(other.name))
            return false;
        if (type != other.type)
            return false;
        return true;
    }

}

Field 实体-->

@NodeEntity
    public class Field implements Comparable<Field> {

        @GraphId private Long id;

        private String name;
        private FieldType fieldType;
        private SourceType sourceType;

        private String logicalName;
        private String dataType;
        private String dataSize;
        private String description;

        private Field() {
            // Empty constructor required as of Neo4j API 2.0.5
        };

        public Field(String name, FieldType fieldType, SourceType sourceType, String logicalName, String dataType, String dataSize, String description) {
            this.name = name;
            this.fieldType = fieldType;
            this.sourceType = sourceType;
            this.logicalName = logicalName;
            this.dataType = dataType;
            this.dataSize = dataSize;
            this.description = description;
        }
        @Relationship(type = "MAPS-TO", direction = Relationship.UNDIRECTED)
        public Set<Field> fields;

        public void mapsTo(Field field) {
            if (fields == null) {
                fields = new HashSet<Field>();
            }
            fields.add(field);
        }

        /* Getter and Setters */

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        public FieldType getFieldType() {
            return fieldType;
        }

        public void setFieldType(FieldType fieldType) {
            this.fieldType = fieldType;
        }

        public SourceType getSourceType() {
            return sourceType;
        }

        public void setSourceType(SourceType sourceType) {
            this.sourceType = sourceType;
        }

        public String getLogicalName() {
            return logicalName;
        }

        public void setLogicalName(String logicalName) {
            this.logicalName = logicalName;
        }

        public String getDataType() {
            return dataType;
        }

        public void setDataType(String dataType) {
            this.dataType = dataType;
        }

        public String getDataSize() {
            return dataSize;
        }

        public void setDataSize(String dataSize) {
            this.dataSize = dataSize;
        }

        public String getDescription() {
            return description;
        }

        public void setDescription(String description) {
            this.description = description;
        }

        public Set<Field> getFields() {
            return fields;
        }

        public void setFields(Set<Field> fields) {
            this.fields = fields;
        }

        @Override
        public int compareTo(Field other) {
            String name = other.getName();
            FieldType fieldType = other.getFieldType();
            SourceType sourceType = other.getSourceType();
            if(this.name.equalsIgnoreCase(name) && this.fieldType.equals(fieldType) && this.sourceType.equals(sourceType))
                return 0;

            return -1;
        }

        @Override
        public int hashCode() {
            final int prime = 31;
            int result = 1;
            result = prime * result + ((fieldType == null) ? 0 : fieldType.hashCode());
            result = prime * result + ((name == null) ? 0 : name.hashCode());
            result = prime * result + ((sourceType == null) ? 0 : sourceType.hashCode());
            return result;
        }

        @Override
        public boolean equals(Object obj) {
            if (this == obj)
                return true;
            if (obj == null)
                return false;
            if (getClass() != obj.getClass())
                return false;
            Field other = (Field) obj;
            if (fieldType != other.fieldType)
                return false;
            if (name == null) {
                if (other.name != null)
                    return false;
            } else if (!name.equals(other.name))
                return false;
            if (sourceType != other.sourceType)
                return false;
            return true;
        }


    }

所以，一个 Source CONTAINS 多个 Field。 Field 是 MAPS-TO 一个或多个其他 Field。

每个 Source 属于 SourceType。

我的不同 SourceType 是：生产者、入站、暂存、中间、出站、消费者。

public enum SourceType {
        PRODUCER, INBOUND, STAGING, INTERMEDIATE, OUTBOUND, CONSUMER;
    }

每个 Field 属于 FieldType。

我的不同 FieldType 是：FILE_FIELD、DB_COLUMN。

public enum FieldType {
    FILE_FIELD, DB_COLUMN;
}

我的数据沿袭如下：生产者 --> 入站 --> 登台 --> 中间 --> 出站 --> 消费者

我现在正在寻找一个高级的 Cypher 查询，如果我在 CONSUMER Source 中提供一个 Field，我就可以追踪它的血统 back 直到制作人 Source.

同样，我也在寻找一个查询，如果我在 PRODUCER Source 中提供 Field，我就可以跟踪它的沿袭 forward 直到消费者 Source.

我已尝试使用 shortestPath 和 neighbors 函数构建查询，但它似乎无法提取我正在寻找的结果。

任何 suggestions/pointers 将不胜感激。

提前致谢！

UPDATE-1

我的数据沿袭背景：我的应用程序从外部应用程序 (PRODUCE) 获取文件。我知道外部应用程序的哪个数据库 tables/columns 填充了文件中的字段。所以在这里，PRODUCER 将是我的 Source 节点；外部应用程序（填充文件）的每个 table.column 都是一个 Field 节点，PRODUCER Source 节点将与所有 Field 节点（代表table.column 填充文件的外部应用程序数据库 table。

来自外部应用程序的文件称为 INBOUND。它是一个逗号分隔的文件。我知道文件中的字段名称是什么以及顺序是什么。所以在这里，INBOUND 将是我的 Source 节点；文件中的每个字段将是一个 Field 节点，INBOUND Source 节点将与所有 Field 节点（表示入站文件中的文件字段）具有 CONTAINS 关系。此外，INBOUND Source 的每个 Field 节点将与 PRODUCER Source 的 Field 节点具有 MAPS_TO 关系（一对一映射）。

继续类似的工作流程，我的下一个阶段称为 STAGING，其中我将入站文件字段加载到我的数据库中 table/column。所以在这里，STAGING 将是我的 Source 节点，数据库 table 的每一列（我将文件字段加载到其中）将代表一个 Field 节点。 STAGING 源节点将与所有 Field 节点（表示我将文件字段加载到的 db table 的 db table.column 有 CONTAINS 关系）。此外，STAGING Source 的每个 Field 节点都将与 INBOUND Source 的 Field 节点具有 MAPS_TO 关系（一对一映射）。

类似我的下一个阶段是中级。在这个阶段，我正在查询 table ，我在其中加载了输入文件的字段，然后将输出刷新到另一个文件中（根据我的业务用例，我可以选择查询全部或仅查询一个子集table 从输入文件填充的列）。我知道哪些字段以何种顺序进入我的中间文件。所以在这里，中间是我的 Source 节点，进入中间文件的每个字段代表我的 Field 节点。此外，INTERMEDIATE Source 将与表示中间文件中字段的所有 Field 节点具有 CONTAINS 关系。此外，这些 Field 个节点中的每一个都将与 STAGING Source 的字段具有 MAPS_TO 关系（一对一映射）。

同样，我有 OUTBOUND 阶段，最后是 CONSUMER 阶段。

...（希望你现在能够观想传承）

我的查询的 objective 是，比如说，如果我给出一个 Field 名称（代表一个 table.column 生产者）作为输入，那么我应该能够跟踪沿着它的血统直到 CONSUMER（即我血统的最后阶段）。

Answer 1

Similarly, I'm also looking for a query by which if I provide a Field in PRODUCER Source, I'm able to track its lineage forward until CONSUMER Source.

我认为我还没有完全理解您的数据模型和要求，但这里有一个关于此查询的想法：

MATCH
  (:Field  {name: { fieldName } })<-[:CONTAINS]-
  (:Source {type: "PRODUCER"    })-[:MAPS_TO]->
  (:Source {type: "INBOUND"     })-[:MAPS_TO]->
  (:Source {type: "STAGING"     })-[:MAPS_TO]->
  (:Source {type: "INTERMEDIATE"})-[:MAPS_TO]->
  (:Source {type: "OUTBOUND"    })-[:MAPS_TO]->
  (consumer:Source {type: "CONSUMER"    })
RETURN consumer

Answer 2

我能够通过以下查询获得所需的数据沿袭：

MATCH (f5:Field)-[:MAPS_TO]-(f4:Field)-[:MAPS_TO]-(f3:Field)-[:MAP‌S_TO]-(f2:Field)-[:M‌APS_TO]-(f1:Field)-[‌:MAPS_TO]-(f:Field)<‌-[:CONTAINS]-(s:Sour‌ce {type: "SOURCE"}) WHERE f.name="<my input source field>" RETURN f,s,f1,f2,f3,f4,f5

有没有办法通过 Neo4j Cypher 查询来跟踪端到端数据沿袭？

Is there a way to track end-to-end data lineage through Neo4j Cypher query?

path

neo4j

cypher

spring-data

data-lineage