4. 数据接入层开发：CSV/JSON数据解析、数据清洗与校验、批量导入图数据库、增量更新策略

数据接入层，说白了就是图计算的「入口」。你想想看，图数据库再牛，没有数据进去也是白搭。我个人习惯把这一层叫做「数据管道」——它负责把原始数据从各种格式里捞出来，洗干净，再塞进图数据库里。

我在项目中遇到过不少坑，比如CSV文件里混进了乱码，JSON嵌套太深导致解析超时，还有批量导入时直接把图数据库搞崩了。嗯，这些经验我会一一拆解给你看。

4.1 CSV/JSON数据解析：别小看这一步

很多人觉得解析CSV和JSON很简单，不就是读文件嘛。但实际生产环境里，数据格式的「脏」程度远超你的想象。

4.1.1 CSV解析的坑与对策

CSV文件看似简单，但有几个常见问题：字段里包含逗号、引号没转义、换行符在字段内部、BOM头干扰。我曾经接手过一个项目，上游系统导出的CSV文件里，某个字段居然包含了完整的HTML代码，逗号和引号满天飞。

我的建议是：不要自己写CSV解析器，用成熟的库。在Java里用OpenCSV或Apache Commons CSV，在Python里用pandas的read_csv。但要注意参数配置：

// Java示例：使用OpenCSV处理带引号的字段
CSVParser parser = new CSVParserBuilder()
    .withSeparator(',')
    .withQuoteChar('"')
    .withEscapeChar('\\')
    .withStrictQuotes(false)
    .build();

CSVReader reader = new CSVReaderBuilder(
    new FileReader("transactions.csv"))
    .withCSVParser(parser)
    .withSkipLines(1)  // 跳过表头
    .build();

String[] line;
while ((line = reader.readNext()) != null) {
    // 每行数据，line数组长度可能不一致
    if (line.length != EXPECTED_COLUMNS) {
        // 记录异常行，不要直接抛异常
        log.warn("行数据列数异常: {}", Arrays.toString(line));
        continue;
    }
    // 处理正常数据
}

4.1.2 JSON解析：扁平化是关键

图数据库需要的是「节点-边」结构，而JSON往往是嵌套的。比如一个用户JSON里嵌套了地址、订单列表。你需要把嵌套结构拆成多个节点和边。

我常用的策略是：先定义JSON Schema，再写解析器。这样解析逻辑清晰，也方便后续校验。

// JSON Schema示例（简化版）
{
  "type": "object",
  "properties": {
    "user_id": {"type": "string"},
    "name": {"type": "string"},
    "address": {
      "type": "object",
      "properties": {
        "city": {"type": "string"},
        "district": {"type": "string"}
      }
    },
    "orders": {
      "type": "array",
      "items": {
        "type": "object",
        "properties": {
          "order_id": {"type": "string"},
          "amount": {"type": "number"}
        }
      }
    }
  }
}

解析时，我会把address拆成「地址节点」，orders拆成「订单节点」，user_id作为「用户节点」，然后用边连接它们。说白了，就是把JSON的树形结构，映射成图的网状结构。

4.2 数据清洗与校验：脏数据是图数据库的噩梦

图数据库对数据质量的要求比关系型数据库更高。为什么？因为一个脏节点可能污染整个图结构。我在项目中见过一个案例：某个节点的ID字段为空，结果批量导入时，图数据库自动生成了空字符串ID，导致后续所有关联查询都查不到数据。

4.2.1 清洗规则：我总结的「三必清」

• 空值必清：节点ID、边起点/终点ID不能为空。其他字段可以允许空值，但要有默认值。
• 格式必清：日期统一成yyyy-MM-dd HH:mm:ss，金额统一成BigDecimal（避免浮点精度问题）。
• 重复必清：同一批数据里，节点ID不能重复。边也不能重复（除非业务允许多重边）。

// 清洗示例：处理日期格式
public String cleanDate(String rawDate) {
    if (rawDate == null || rawDate.trim().isEmpty()) {
        return null;  // 或者返回默认值
    }
    // 尝试多种常见格式
    String[] patterns = {"yyyy-MM-dd", "yyyy/MM/dd", "yyyyMMdd", "yyyy-MM-dd HH:mm:ss"};
    for (String pattern : patterns) {
        try {
            SimpleDateFormat sdf = new SimpleDateFormat(pattern);
            Date date = sdf.parse(rawDate.trim());
            // 统一输出格式
            return new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(date);
        } catch (ParseException e) {
            // 继续尝试下一种格式
        }
    }
    // 所有格式都失败，记录异常
    log.error("无法解析的日期格式: {}", rawDate);
    return null;  // 或者抛出异常
}

4.2.2 校验：不止是「非空检查」

校验要分两层：结构校验和业务校验。

• 结构校验：字段类型对不对？节点ID是否唯一？边的起点和终点是否存在？
• 业务校验：金额不能为负数？用户年龄不能超过150？交易时间不能晚于当前时间？

4.3 批量导入图数据库：性能与稳定性的平衡

批量导入是数据接入层的核心环节。我见过有人用单条INSERT语句插入百万级数据，结果跑了三天三夜还没跑完。也见过有人一次性提交10万条数据，直接把图数据库的内存打爆。

4.3.1 批量导入的策略

以Neo4j为例，我推荐使用BATCH模式，每批500-2000条数据。这个区间是我反复测试得出的经验值——太少则网络开销大，太多则事务压力大。

// Neo4j批量导入示例（使用Java Driver）
try (Session session = driver.session()) {
    // 每批1000条
    int batchSize = 1000;
    List<Map<String, Object>> batch = new ArrayList<>();
    
    for (Map<String, Object> record : allRecords) {
        batch.add(record);
        if (batch.size() >= batchSize) {
            session.writeTransaction(tx -> {
                for (Map<String, Object> data : batch) {
                    tx.run("CREATE (n:User {id: $id, name: $name})", data);
                }
                return null;
            });
            batch.clear();
        }
    }
    // 处理剩余数据
    if (!batch.isEmpty()) {
        session.writeTransaction(tx -> {
            for (Map<String, Object> data : batch) {
                tx.run("CREATE (n:User {id: $id, name: $name})", data);
            }
            return null;
        });
    }
}

4.3.2 导入失败的处理

批量导入最怕中途失败。我的做法是：记录断点。每成功导入一批，就把这批的「最后一条记录ID」写入一个状态文件或数据库表。如果程序崩溃重启，从断点处继续导入，而不是从头开始。

4.4 增量更新策略：别每次都全量导入

全量导入只适合第一次上线。后续的数据更新，必须用增量策略。否则数据量大了以后，全量导入一次可能要跑几个小时，业务根本等不起。

4.4.1 常见的增量更新方式

方式	原理	适用场景	我的评价
时间戳增量	根据数据更新时间字段，只拉取最近变更的数据	数据源有明确的更新时间字段	最简单，但依赖上游数据质量
日志捕获（CDC）	监听数据库的binlog或WAL日志，实时捕获变更	需要实时性高的场景	技术门槛高，但效果最好
消息队列	上游系统将变更事件发送到Kafka/RocketMQ，下游消费	微服务架构，事件驱动	我最推荐的方式，解耦性好
全量对比	定期全量拉取，与图数据库中的数据进行比对，找出差异	数据量不大，且无法获取增量信息	简单粗暴，但效率低

4.4.2 增量更新的核心原则：幂等性

增量更新操作必须是幂等的。什么意思？就是同样的数据执行多次，结果是一样的。比如「MERGE」操作（有则更新，无则创建）就是幂等的，而「CREATE」操作不是。

// 幂等的增量更新示例（Cypher）
MERGE (n:User {id: $id})
SET n.name = $name, n.update_time = $updateTime

// 边的增量更新
MATCH (a:User {id: $fromId})
MATCH (b:User {id: $toId})
MERGE (a)-[r:FOLLOWS]->(b)
SET r.update_time = $updateTime

4.5 本章知识体系总览

下面这张图，是我梳理的数据接入层核心流程。你可以把它当作开发时的「路线图」。

这张图把整个数据接入流程串起来了。从数据源开始，经过解析、清洗校验，然后根据场景选择全量或增量策略，最后批量写入图数据库。异常数据走死信队列，不影响主流程。

我个人觉得，数据接入层开发最考验的不是技术能力，而是对数据质量的敬畏心。你想想看，图数据库里的数据一旦脏了，清洗成本极高。所以，宁可多花时间在清洗和校验上，也不要图省事直接导入。

---