数据导入与知识图谱构建流程

概述

在通过 .schema 文件建立好实体模型后，用户导入数据并建立知识图谱的完整流程如下:

.schema 文件 → Schema解析 → MySQL元数据 → Neo4j约束/索引
                                    ↓
CSV/JSON数据 → Pipeline流水线 → BuilderRecord → SPGRecord → LPG Record → Neo4j图数据

本文档重点介绍数据导入阶段，即如何将CSV/JSON格式的原始数据转换为Neo4j知识图谱。

整体架构

Schema建模阶段 (已完成)

上传 .schema 文件 → SchemaController.uploadSchema()
解析DSL → SchemaScriptTranslateUtil.translateScript()
保存到MySQL → SchemaServiceImpl.saveSchema()
- kg_ontology_entity - 实体类型定义
- kg_ontology_property_range - 属性和关系定义
- kg_property_constraint - 约束规则
- kg_project_entity - 项目实体关联
- kg_ontology_release - 版本管理
同步到Neo4j → SchemaGraphSyncService.syncSchemaToNeo4j()
- 创建唯一性约束: CREATE CONSTRAINT uniqueness_TypeName_id FOR (n:TypeName) REQUIRE n.id IS UNIQUE
- 创建属性索引: CREATE INDEX index_TypeName_propertyName FOR (n:TypeName) ON (n.propertyName)

数据导入阶段 (本文重点)

原始数据 (CSV/JSON)
    ↓
Pipeline流水线
    ↓
BuilderRecord (构建记录)
    ↓
SPGRecord (SPG语义记录)
    ├── EntityRecord (实体记录)
    └── RelationRecord (关系记录)
    ↓
LPG Record (标签属性图记录)
    ├── VertexRecord (顶点记录)
    └── EdgeRecord (边记录)
    ↓
Neo4j Cypher执行
    ↓
知识图谱建立完成

Pipeline 流水线架构

Pipeline模型

Pipeline采用**有向无环图(DAG)**结构，由节点(Node)和边(Edge)组成:

java

public class Pipeline implements Serializable {
    private List<PipelineNode> nodes;  // 节点列表
    private List<PipelineEdge> edges;  // 边列表
}

典型Pipeline结构

┌─────────────┐
│ CSV_SOURCE  │  ← 数据源节点 (读取CSV/JSON文件)
└──────┬──────┘
       │
       ↓
┌──────────────────┐
│ SPG_TYPE_MAPPING │  ← 类型映射节点 (BuilderRecord → SPGRecord)
└──────┬───────────┘
       │
       ↓
┌─────────────┐
│ NEO4J_SINK  │  ← 数据写入节点 (SPGRecord → Neo4j)
└─────────────┘

节点类型

节点类型	实现类	功能
CSV_SOURCE	`CsvFileSourceReader`	读取CSV文件为BuilderRecord
SPG_TYPE_MAPPING	`SPGTypeMappingProcessor`	将BuilderRecord转换为SPGRecord
NEO4J_SINK	`Neo4jSinkWriter`	将SPGRecord写入Neo4j
EXTRACT	`RelationExtractProcessor`	关系抽取(可选)
FUSE	`EntityFuseProcessor`	实体融合(可选)

数据转换流程

第一阶段: CSV → BuilderRecord

输入: CSV文件

csv

id,name,age,city
1001,张三,28,北京
1002,李四,35,上海

输出: BuilderRecord

java

public class BuilderRecord extends BaseRecord {
    private Map<String, Object> properties;  // 原始CSV字段映射
}

示例:

java

BuilderRecord {
    properties: {
        "id": "1001",
        "name": "张三",
        "age": 28,
        "city": "北京"
    }
}

核心代码: CsvFileSourceReader.java:69-86

第二阶段: BuilderRecord → SPGRecord

处理器: SPGTypeMappingProcessor

转换逻辑 (SPGTypeMappingHelper.java:52-116):

获取SPG类型定义 (从Schema元数据)

java

SPGTypeIdentifier spgTypeIdentifier = mapping.getSpgTypeIdentifier();
BaseSpgType spgType = projectSchema.getSpgType(spgTypeIdentifier);

提取主键

java

String bizId = SPGRecordAlterUtils.getPropertyValue(record, idMapping);

映射属性 (遍历Schema定义的所有属性)

java

for (BaseProperty property : spgType.getAllProperties()) {
    Object value = SPGRecordAlterUtils.getPropertyValue(record, propertyMapping);
    properties.put(property.getName(), value);
}

创建SPGRecord
- 实体类型 → EntityRecord
- 关系类型 → RelationRecord

输出示例:

java

// 实体记录
EntityRecord {
    id: "Person_1001",
    name: "Person",
    properties: {
        "id": "1001",
        "name": "张三",
        "age": 28,
        "city": "北京"
    }
}

// 关系记录
RelationRecord {
    srcId: "Person_1001",
    dstId: "City_北京",
    name: "livesIn",
    properties: {
        "since": "2020-01-01"
    }
}

核心文件:

SPGTypeMappingProcessor.java:52-85 - 主处理器
SPGTypeMappingHelper.java:52-116 - 映射逻辑
SPGRecordTypeEnum.java - 记录类型枚举

第三阶段: SPGRecord → LPG Record

转换器:

VertexRecordConvertor - 实体 → 顶点
EdgeRecordConvertor - 关系 → 边

转换逻辑 (VertexRecordConvertor.java:33-76):

提取节点ID
java
```
String vertexId = spgRecord.getId();
```
提取Label (节点类型)
java
```
String label = spgRecord.getName();
```

转换属性 (过滤掉关系属性)

java

Map<String, Object> properties = new HashMap<>(spgRecord.getProperties());
// 移除非本地属性
for (BaseProperty property : spgType.getAllProperties()) {
    if (!property.isSemanticProperty()) {
        properties.remove(property.getName());
    }
}

创建VertexRecord

java

return new VertexRecord(vertexId, label, properties);

输出示例:

java

// 顶点记录
VertexRecord {
    id: "Person_1001",
    label: "Person",
    properties: {
        "id": "1001",
        "name": "张三",
        "age": 28,
        "city": "北京"
    }
}

// 边记录
EdgeRecord {
    srcId: "Person_1001",
    dstId: "City_北京",
    type: "livesIn",
    properties: {
        "since": "2020-01-01"
    }
}

核心文件:

VertexRecordConvertor.java:33-76 - 顶点转换
EdgeRecordConvertor.java:34-73 - 边转换
LPGTypeEnum.java - LPG类型枚举

第四阶段: LPG Record → Neo4j Cypher

写入器: Neo4jSinkWriter

Cypher生成逻辑 (Neo4jSinkWriter.java:77-164):

顶点写入

java

private void writeVertex(VertexRecord vertex) {
    String cypher = String.format(
        "MERGE (n:%s {id: $id}) SET n += $properties",
        vertex.getLabel()
    );

    Map<String, Object> params = Map.of(
        "id", vertex.getId(),
        "properties", vertex.getProperties()
    );

    session.executeWrite(tx -> {
        tx.run(cypher, params);
        return null;
    });
}

生成的Cypher:

cypher

MERGE (n:Person {id: 'Person_1001'})
SET n += {
    id: '1001',
    name: '张三',
    age: 28,
    city: '北京'
}

边写入

java

private void writeEdge(EdgeRecord edge) {
    String cypher = String.format(
        "MATCH (a {id: $srcId}), (b {id: $dstId}) " +
        "MERGE (a)-[r:%s]->(b) " +
        "SET r += $properties",
        edge.getType()
    );

    Map<String, Object> params = Map.of(
        "srcId", edge.getSrcId(),
        "dstId", edge.getDstId(),
        "properties", edge.getProperties()
    );

    session.executeWrite(tx -> {
        tx.run(cypher, params);
        return null;
    });
}

生成的Cypher:

cypher

MATCH (a {id: 'Person_1001'}), (b {id: 'City_北京'})
MERGE (a)-[r:livesIn]->(b)
SET r += {since: '2020-01-01'}

批量写入 (Neo4jSinkWriter.java:101-123):

java

public void write(Iterator<SPGRecord> records) {
    List<SPGRecord> batch = new ArrayList<>();

    while (records.hasNext()) {
        batch.add(records.next());

        if (batch.size() >= batchSize) {
            writeBatch(batch);
            batch.clear();
        }
    }

    if (!batch.isEmpty()) {
        writeBatch(batch);
    }
}

核心组件详解

1. LocalBuilderRunner - Pipeline执行引擎

位置: LocalBuilderRunner.java:47-125

核心功能:

解析Pipeline DAG结构
按拓扑顺序执行节点
管理节点间数据流
支持并行执行

执行流程:

java

public void run(Pipeline pipeline) {
    // 1. 构建执行计划
    List<List<PipelineNode>> layers = topologicalSort(pipeline);

    // 2. 按层执行
    for (List<PipelineNode> layer : layers) {
        // 并行执行同一层的节点
        layer.parallelStream().forEach(node -> {
            // 3. 获取上游数据
            Iterator<BaseRecord> input = getInputData(node);

            // 4. 执行节点处理
            Iterator<BaseRecord> output = node.getProcessor().process(input);

            // 5. 传递给下游
            passToDownstream(node, output);
        });
    }
}

拓扑排序示例:

层级0: [CSV_SOURCE]
层级1: [SPG_TYPE_MAPPING]
层级2: [NEO4J_SINK]

2. SPGTypeMappingProcessor - 类型映射处理器

位置: SPGTypeMappingProcessor.java:52-85

核心职责:

加载项目Schema定义
根据Mapping配置转换BuilderRecord
生成EntityRecord/RelationRecord
处理属性映射和类型转换

配置示例:

json

{
    "spgTypeIdentifier": {
        "namespace": "medical",
        "name": "Person"
    },
    "idMapping": {
        "sourceName": "id"
    },
    "propertyMappings": [
        {
            "targetName": "name",
            "sourceName": "person_name"
        },
        {
            "targetName": "age",
            "sourceName": "person_age",
            "transform": "toInt"
        }
    ]
}

3. Neo4jSinkWriter - Neo4j批量写入器

位置: Neo4jSinkWriter.java:77-164

性能优化:

批量写入 (默认1000条/批次)

java

private static final int DEFAULT_BATCH_SIZE = 1000;

事务管理

java

session.executeWrite(tx -> {
    for (SPGRecord record : batch) {
        String cypher = generateCypher(record);
        tx.run(cypher, params);
    }
    return null;
});

并行写入 (多线程)

java

ExecutorService executor = Executors.newFixedThreadPool(8);
batches.forEach(batch -> executor.submit(() -> writeBatch(batch)));

失败重试

java

@Retryable(
    value = {Neo4jException.class},
    maxAttempts = 3,
    backoff = @Backoff(delay = 1000)
)
public void writeBatch(List<SPGRecord> batch) {
    // ...
}

代码示例

示例1: 创建简单Pipeline

java

// 1. 定义数据源节点
PipelineNode sourceNode = new PipelineNode();
sourceNode.setId("csv_source");
sourceNode.setNodeType(NodeTypeEnum.CSV_SOURCE);
sourceNode.setConfig(Map.of(
    "filePath", "/data/person.csv",
    "delimiter", ","
));

// 2. 定义映射节点
PipelineNode mappingNode = new PipelineNode();
mappingNode.setId("type_mapping");
mappingNode.setNodeType(NodeTypeEnum.SPG_TYPE_MAPPING);
mappingNode.setConfig(Map.of(
    "spgType", "medical.Person",
    "idMapping", Map.of("sourceName", "id"),
    "propertyMappings", List.of(
        Map.of("targetName", "name", "sourceName", "person_name"),
        Map.of("targetName", "age", "sourceName", "person_age")
    )
));

// 3. 定义写入节点
PipelineNode sinkNode = new PipelineNode();
sinkNode.setId("neo4j_sink");
sinkNode.setNodeType(NodeTypeEnum.NEO4J_SINK);
sinkNode.setConfig(Map.of(
    "batchSize", 1000,
    "parallelism", 8
));

// 4. 定义边
PipelineEdge edge1 = new PipelineEdge("csv_source", "type_mapping");
PipelineEdge edge2 = new PipelineEdge("type_mapping", "neo4j_sink");

// 5. 组装Pipeline
Pipeline pipeline = new Pipeline();
pipeline.setNodes(List.of(sourceNode, mappingNode, sinkNode));
pipeline.setEdges(List.of(edge1, edge2));

// 6. 执行
LocalBuilderRunner runner = new LocalBuilderRunner();
runner.run(pipeline);

示例2: CSV数据导入完整流程

CSV文件 (person.csv):

csv

id,person_name,person_age,city_name
1001,张三,28,北京
1002,李四,35,上海
1003,王五,42,深圳

Schema定义 (已通过 .schema 文件上传):

namespace medical

Person(人员) {
    id: Text
    name: Text
    age: Integer
    livesIn: City
}

City(城市) {
    id: Text
    name: Text
}

Pipeline配置:

json

{
    "nodes": [
        {
            "id": "person_source",
            "nodeType": "CSV_SOURCE",
            "config": {
                "filePath": "/data/person.csv",
                "delimiter": ",",
                "hasHeader": true
            }
        },
        {
            "id": "person_mapping",
            "nodeType": "SPG_TYPE_MAPPING",
            "config": {
                "spgTypeIdentifier": {
                    "namespace": "medical",
                    "name": "Person"
                },
                "idMapping": {
                    "sourceName": "id"
                },
                "propertyMappings": [
                    {
                        "targetName": "name",
                        "sourceName": "person_name"
                    },
                    {
                        "targetName": "age",
                        "sourceName": "person_age",
                        "transform": "toInt"
                    },
                    {
                        "targetName": "livesIn",
                        "sourceName": "city_name",
                        "objectTypeRef": "City"
                    }
                ]
            }
        },
        {
            "id": "neo4j_sink",
            "nodeType": "NEO4J_SINK",
            "config": {
                "batchSize": 1000
            }
        }
    ],
    "edges": [
        {"from": "person_source", "to": "person_mapping"},
        {"from": "person_mapping", "to": "neo4j_sink"}
    ]
}

执行后的Neo4j图谱:

cypher

// 节点
(:Person {id: '1001', name: '张三', age: 28})
(:Person {id: '1002', name: '李四', age: 35})
(:Person {id: '1003', name: '王五', age: 42})
(:City {id: 'City_北京', name: '北京'})
(:City {id: 'City_上海', name: '上海'})
(:City {id: 'City_深圳', name: '深圳'})

// 关系
(:Person {id: '1001'})-[:livesIn]->(:City {name: '北京'})
(:Person {id: '1002'})-[:livesIn]->(:City {name: '上海'})
(:Person {id: '1003'})-[:livesIn]->(:City {name: '深圳'})

示例3: 关系数据导入

CSV文件 (friendship.csv):

csv

person_a_id,person_b_id,since_year,relation_type
1001,1002,2018,朋友
1002,1003,2020,同事
1001,1003,2021,校友

Schema定义:

namespace medical

Person(人员) {
    id: Text
    name: Text
    knows: Person
}

knows(认识关系) {
    since: Integer
    relationType: Text
}

Pipeline配置:

json

{
    "nodes": [
        {
            "id": "relation_source",
            "nodeType": "CSV_SOURCE",
            "config": {
                "filePath": "/data/friendship.csv"
            }
        },
        {
            "id": "relation_mapping",
            "nodeType": "SPG_TYPE_MAPPING",
            "config": {
                "spgTypeIdentifier": {
                    "namespace": "medical",
                    "name": "knows"
                },
                "subjectMapping": {
                    "sourceName": "person_a_id",
                    "objectType": "Person"
                },
                "objectMapping": {
                    "sourceName": "person_b_id",
                    "objectType": "Person"
                },
                "propertyMappings": [
                    {
                        "targetName": "since",
                        "sourceName": "since_year",
                        "transform": "toInt"
                    },
                    {
                        "targetName": "relationType",
                        "sourceName": "relation_type"
                    }
                ]
            }
        },
        {
            "id": "neo4j_sink",
            "nodeType": "NEO4J_SINK"
        }
    ],
    "edges": [
        {"from": "relation_source", "to": "relation_mapping"},
        {"from": "relation_mapping", "to": "neo4j_sink"}
    ]
}

执行后的Cypher:

cypher

MATCH (a:Person {id: '1001'}), (b:Person {id: '1002'})
MERGE (a)-[r:knows]->(b)
SET r += {since: 2018, relationType: '朋友'}

MATCH (a:Person {id: '1002'}), (b:Person {id: '1003'})
MERGE (a)-[r:knows]->(b)
SET r += {since: 2020, relationType: '同事'}

MATCH (a:Person {id: '1001'}), (b:Person {id: '1003'})
MERGE (a)-[r:knows]->(b)
SET r += {since: 2021, relationType: '校友'}

文件索引

Pipeline执行引擎

文件路径	核心功能	关键方法
`LocalBuilderRunner.java:47-125`	Pipeline执行引擎	`run()`, `topologicalSort()`
`Pipeline.java:28-45`	Pipeline模型定义	`getNodes()`, `getEdges()`
`PipelineNode.java:23-52`	Pipeline节点定义	`getNodeType()`, `getConfig()`
`PipelineEdge.java:18-32`	Pipeline边定义	`getFrom()`, `getTo()`

数据源节点

文件路径	核心功能	关键方法
`CsvFileSourceReader.java:69-86`	CSV文件读取	`read()`
`JsonFileSourceReader.java:54-78`	JSON文件读取	`read()`
`BuilderRecord.java:25-48`	构建记录模型	`getProperties()`

类型映射节点

文件路径	核心功能	关键方法
`SPGTypeMappingProcessor.java:52-85`	类型映射处理器	`process()`
`SPGTypeMappingHelper.java:52-116`	映射逻辑实现	`toSPGRecord()`
`SPGRecordAlterUtils.java:34-89`	记录转换工具	`getPropertyValue()`

记录模型

文件路径	核心功能	关键方法
`SPGRecord.java:30-65`	SPG记录基类	`getId()`, `getName()`
`EntityRecord.java:22-45`	实体记录	`getProperties()`
`RelationRecord.java:25-58`	关系记录	`getSrcId()`, `getDstId()`
`SPGRecordTypeEnum.java:18-28`	记录类型枚举	`ENTITY`, `RELATION`

图记录转换

文件路径	核心功能	关键方法
`VertexRecordConvertor.java:33-76`	SPG→顶点转换	`toVertexRecord()`
`EdgeRecordConvertor.java:34-73`	SPG→边转换	`toEdgeRecord()`
`VertexRecord.java:21-42`	顶点记录模型	`getLabel()`
`EdgeRecord.java:23-51`	边记录模型	`getType()`
`LPGTypeEnum.java:17-25`	LPG类型枚举	`VERTEX`, `EDGE`

Neo4j写入节点

文件路径	核心功能	关键方法
`Neo4jSinkWriter.java:77-164`	Neo4j批量写入	`write()`, `writeBatch()`
`CypherGenerator.java:45-128`	Cypher语句生成	`generateVertexCypher()`, `generateEdgeCypher()`

Schema服务

文件路径	核心功能	关键方法
`SchemaServiceImpl.java:102-147`	Schema上传解析	`uploadSchema()`, `parseSchema()`
`SchemaGraphSyncService.java:49-125`	Neo4j Schema同步	`syncSchemaToNeo4j()`
`GraphServiceImpl.java:64-240`	图数据CRUD	`createEntity()`, `createRelation()`

Mapper (数据库)

文件路径	核心功能	SQL操作
`KgOntologyEntityMapper.java`	实体类型表操作	`insert()`, `selectByUniqueName()`
`KgOntologyPropertyRangeMapper.java`	属性关系表操作	`insert()`
`KgPropertyConstraintMapper.java`	约束表操作	`insert()`
`KgProjectEntityMapper.java`	项目实体关联表操作	`insert()`
`KgOntologyReleaseMapper.java`	Schema版本表操作	`insert()`, `selectOnlineVersionByProjectId()`

总结

整个知识图谱构建流程分为两个主要阶段:

阶段一: Schema建模

用户上传 .schema 文件
系统解析DSL并保存到MySQL元数据库
系统在Neo4j中创建约束和索引

阶段二: 数据导入

CSV_SOURCE: 读取CSV文件为BuilderRecord
SPG_TYPE_MAPPING: 根据Schema定义转换为SPGRecord (EntityRecord/RelationRecord)
LPG转换: 将SPGRecord转换为VertexRecord/EdgeRecord
NEO4J_SINK: 批量生成Cypher并写入Neo4j

核心设计模式:

Pipeline模式: DAG结构支持复杂数据流
分层转换: BuilderRecord → SPGRecord → LPG Record → Neo4j
批量处理: 1000条/批次,支持并行执行
Schema驱动: 所有映射基于Schema元数据

性能优化:

批量写入 (默认1000条)
并行执行 (多线程)
事务管理
失败重试

文档版本: v1.0 最后更新: 2025-12-21 维护者: Aroundix SPG Team

数据导入与知识图谱构建流程 ​

目录 ​

概述 ​

整体架构 ​

Schema建模阶段 (已完成) ​

数据导入阶段 (本文重点) ​

Pipeline 流水线架构 ​

Pipeline模型 ​

典型Pipeline结构 ​

节点类型 ​

数据转换流程 ​

第一阶段: CSV → BuilderRecord ​

第二阶段: BuilderRecord → SPGRecord ​

第三阶段: SPGRecord → LPG Record ​

第四阶段: LPG Record → Neo4j Cypher ​

顶点写入 ​

边写入 ​

核心组件详解 ​

1. LocalBuilderRunner - Pipeline执行引擎 ​

2. SPGTypeMappingProcessor - 类型映射处理器 ​

3. Neo4jSinkWriter - Neo4j批量写入器 ​

代码示例 ​

示例1: 创建简单Pipeline ​

示例2: CSV数据导入完整流程 ​

示例3: 关系数据导入 ​

文件索引 ​

Pipeline执行引擎 ​

数据源节点 ​

类型映射节点 ​

记录模型 ​

图记录转换 ​

Neo4j写入节点 ​

Schema服务 ​

Mapper (数据库) ​

总结 ​

阶段一: Schema建模 ​

阶段二: 数据导入 ​

数据导入与知识图谱构建流程

目录

概述

整体架构

Schema建模阶段 (已完成)

数据导入阶段 (本文重点)

Pipeline 流水线架构

Pipeline模型

典型Pipeline结构

节点类型

数据转换流程

第一阶段: CSV → BuilderRecord

第二阶段: BuilderRecord → SPGRecord

第三阶段: SPGRecord → LPG Record

第四阶段: LPG Record → Neo4j Cypher

顶点写入

边写入

核心组件详解

1. LocalBuilderRunner - Pipeline执行引擎

2. SPGTypeMappingProcessor - 类型映射处理器

3. Neo4jSinkWriter - Neo4j批量写入器

代码示例

示例1: 创建简单Pipeline

示例2: CSV数据导入完整流程

示例3: 关系数据导入

文件索引

Pipeline执行引擎

数据源节点

类型映射节点

记录模型

图记录转换

Neo4j写入节点

Schema服务

Mapper (数据库)

总结

阶段一: Schema建模

阶段二: 数据导入