自定义过程与 APOC：Neo4j 过程库与 Java 扩展

Cypher 能：声明式查询、模式匹配、聚合、子查询
Cypher 做不了：
  - 调用外部服务（HTTP/RPC/数据库）
  - 复杂迭代算法（依赖命令式控制流）
  - 批量生成/变换大量数据（过程式）
  - 解析非结构化文本（需通用编程）
  - 复用复杂逻辑（Cypher 无「函数库」分发）
→ 查询语言适合「声明查询」，不适合「命令式/集成」

- 过程（Procedure）：返回「多行结果表」，可执行操作
  CALL apoc.export.csv.query(query, file)  → 导出
- 函数（Function）：返回「单个值」，用在表达式里
  apoc.text.clean(text)  → 清洗文本
→ 过程 = 多行操作，函数 = 单值计算

- APOC 是社区标准库（300+ 过程/函数）
- 覆盖：文本/图/路径/导入导出/并发/地理
- 开箱即用：装插件即可调用
- 避免重复造轮子（先查 APOC 有没有现成的）
→ 原则：先 APOC，不够再自写

- APOC 没有的功能（特定业务算法）
- 性能敏感（Cypher 循环太慢 → Java 原生）
- 需要外部集成（第三方 API/SDK）
- 需要复用（团队共享的「领域过程库」）
→ 自写 = APOC 覆盖不到 + 性能/集成/复用需求

- 图操作：subgraph、paths（路径展开）、index
- 文本处理：clean/format/join/replace/regex
- 集合/列表：flatten/sort/min/max（内存操作）
- 导入导出：csv/json/load/export
- 并发：periodic.iterate（批量并行处理）
- 触发：triggers（写事件钩子）
- 地理/空间：geo 距离/坐标转换
- 元数据：schema/index/constraint 查询
→ APOC = 一张「常用图操作」的现成工具表

- apoc.xxx 前缀：核心功能（随版本稳定）
- apoc.path/apoc.text/apoc.export 等分类命名
- 版本对齐：APOC 版本与 Neo4j 版本匹配
  （apoc-5.x 配 neo4j-5.x）
- 部分功能需「扩展 jar」（apoc-extra/extended）
→ 用之前查文档确认类别与版本

- GDS（Graph Data Science）是独立的算法过程库
- 覆盖：中心性/社区/相似度/嵌入等算法
- 调用方式一致：CALL gds.xxx.stream()
- 分工：APOC 管「通用操作」，GDS 管「图算法」
→ 图算法需求 → GDS，通用操作 → APOC

src/main/java/com/example/procs/
  └── MyProcedures.java

pom.xml 依赖：
  org.neo4j:neo4j:5.x (provided)
  org.neo4j:procedure-api:5.x
→ 用 procedure-api 编译，运行时由 Neo4j 提供

package com.example.procs;

import org.neo4j.procedure.*;
import java.util.stream.Stream;

public class MyProcedures {

    // 声明为过程，指定名字空间
    @Procedure(name = "example.hello")
    public Stream<HelloResult> hello() {
        return Stream.of(new HelloResult("Hello from Neo4j"));
    }

    // 返回类型的「一行」定义（Public 类 + 公共字段）
    public static class HelloResult {
        public final String greeting;
        public HelloResult(String greeting) {
            this.greeting = greeting;
        }
    }
}

// @Procedure：过程（多行返回）
@Procedure(name = "example.hello", mode = Procedure.WRITE)
public Stream<HelloResult> hello(@Name("name") String name) {
    // mode：READ（默认只读）/ WRITE（可写图）
}

// @UserFunction：函数（单值返回）
@UserFunction("example.clean")
public String clean(@Name("text") String text) {
    return text.trim();
}

// @Name 注解：参数名（Cypher 侧可见）
// 类型：String/Long/Double/Boolean/List/Map/Node/Relationship/Path

mvn package → target/my-procedures-1.0.jar
- 打包后放入 Neo4j plugins 目录
- 重启或热加载（第 7 节）
- 用 CALL 验证（dbms.procedures() 检查）
→ 打包 jar → 放 plugins → 重启/热加载 → 调用验证

- @Procedure 注解 + 返回 Stream<结果行>
- 结果行是「Public 类 + public 字段」
- 参数用 @Name 标注
- 包名用反域名（com.example 避免冲突）
- 类名避免与内置过程冲突
→ 骨架固定：注解 + 返回类 + 参数名

Java 类型         Cypher 类型
String            STRING
long / Long       INTEGER
double / Double   FLOAT
boolean           BOOLEAN
List<T>           LIST
Map<String, T>    MAP
Node              NODE
Relationship      RELATIONSHIP
Path              PATH
→ 参数类型与 Cypher 类型一一对应

// 方式 1：Public 类 + public 字段
public static class CountResult {
    public final String label;
    public final long count;
    public CountResult(String label, long count) {
        this.label = label; this.count = count;
    }
}

// 方式 2：Map 返回（无需自定义类）
@Procedure("example.counts")
public Stream<Map<String, Object>> counts() {
    return Stream.of(Map.of("label", "Person", "count", 10L));
}

- 过程返回 Stream<T>：按需生成行（惰性）
- 适合大结果（不一次性全建）
- 用 Stream.generate/map/filter 组合
- 不要先全构建 List 再转 Stream（失惰性）
→ Stream 返回 = 惰性生成，控制内存

// 注入 GraphDatabaseService（过程上下文）
@Context
public GraphDatabaseService db;

@Procedure("example.degree")
public Stream<DegreeResult> degree(@Name("id") String id) {
    try (Transaction tx = db.beginTx()) {
        Node n = tx.findNode(
            Label.label("Account"), "id", id);
        long degree = n.getDegree(
            RelationshipType.withName("TRANSFER"));
        tx.commit();
        return Stream.of(new DegreeResult(id, degree));
    }
}

- @Context GraphDatabaseService：访问图/事务
- @Context Log：日志
- @Context TermGuard / ProcedureTransaction：事务控制
- 注入在方法字段上（非参数）
→ @Context 让过程能「读写图 + 记录日志 + 控事务」

// 抛异常 → Cypher 侧显示为过程错误
throw new RuntimeException("Invalid id: " + id);

// 用 QueryExecutionException 传递更友好错误
// 过程内部分失败：考虑返回错误行而非中断全部
→ 错误策略：抛异常中断 vs 返回错误标记

// 标量函数：返回单值，用在表达式里
@UserFunction("example.clean")
public String clean(@Name("text") String text) {
    return text == null ? null : text.trim();
}

@UserFunction("example.ratio")
public double ratio(@Name("a") long a, @Name("b") long b) {
    return b == 0 ? 0.0 : (double) a / b;
}

- 函数：返回单值，用在 RETURN/WHERE/SET 表达式
  例：RETURN example.ratio(x, y)
- 过程：返回多行，用 CALL
  例：CALL example.list() YIELD ...
→ 单值计算用函数，多行/操作用过程

// 聚合函数：接受多行输入，输出一个值
@UserAggregationFunction("example.median")
public class Median {
    private List<Double> values = new ArrayList<>();

    // 每行调用一次
    @UserAggregationUpdate
    public void update(@Name("value") Double value) {
        if (value != null) values.add(value);
    }

    // 最终结果
    @UserAggregationResult
    public Double result() {
        if (values.isEmpty()) return null;
        Collections.sort(values);
        int n = values.size();
        return n % 2 == 0
            ? (values.get(n/2-1) + values.get(n/2)) / 2.0
            : values.get(n/2);
    }
}

- @UserAggregationFunction 标注类
- update 方法：每行更新状态（累加/收集）
- result 方法：输出最终聚合值
- 状态在类字段中（一次查询一个实例）
→ 聚合 = 状态收集（update）+ 最终输出（result）

- 函数内不能写图（纯计算/只读）
- 避免副作用（函数应纯（无外部状态））
- 大函数/复杂逻辑 → 用过程更合适
- 函数命名：点分命名空间（example.clean）
→ 函数保持「纯计算」，复杂/写操作用过程

- YIELD：选择过程返回的列（改名/过滤）
- CALL + RETURN：整个过程结果作为临时表
- 子查询：CALL 在 WITH 之后（过程处理上游数据）
- UNWIND：把过程返回列表展开成行
→ 过程结果像「临时表」参与查询

- READ 过程：在事务内执行，可读取当前事务数据
- WRITE 过程：写操作在事务内（可回滚）
- 过程内开的事务与查询事务的关系：
  - 用 db.beginTx() 是新事务（注意嵌套）
  - 推荐用 @Context TransactionManager 管理
→ 事务语义：过程内写入随查询事务提交/回滚

- 过程做「复杂算子」，Cypher 做「声明逻辑」
  例：Cypher 收集输入 → CALL 过程处理 → Cypher 消化
- 过程做「外部集成」，Cypher 做「结果整合」
- 过程做「批量操作」，Cypher 做「批次选择」
→ 分工：Cypher 声明「做什么」，过程实现「怎么做」

- CALL dbms.procedures() 确认注册
- 直接 CALL 测试单过程（入参 → 看输出）
- 与 Cypher 组合测（数据流是否正确）
- 集成测试：JUnit + Neo4j Test Harness
→ 测试：注册确认 → 单过程 → 组合 → 集成

1. 打包 jar（mvn package）
2. 放入 $NEO4J_HOME/plugins
3. 检查配置：dbms.security.procedures.allowlist
4. 重启 Neo4j（或热加载，见下）
5. 验证：CALL dbms.procedures() 找到自己的过程
→ 打包 → plugins → 白名单 → 重启 → 验证

- 插件目录：plugins/（默认扫描 jar）
- 白名单配置：
  dbms.security.procedures.allowlist=example.*,apoc.*
  （未在白名单的过程不可调用——安全）
- 默认内置：dbms.procedures.* 内置可用
→ 白名单是「安全护栏」：只有允许的过程能跑

- Neo4j 支持插件热加载：放新 jar 即生效（部分版本）
- 5.x：默认扫描 plugins 目录，替换 jar 自动加载
- 无热加载时：需重启（或调 dbms.allow_... ）
- 版本差异：检查所使用 Neo4j 版本的插件机制
→ 热加载 = 更新 jar 即生效，无则重启

- 编译用与运行 Neo4j 相同版本（procedure-api）
- 大版本不兼容（4.x 过程在 5.x 需重编译）
- API 演进：@Procedure/返回类稳定，内部 API 会变
- 依赖冲突：避免打包重复依赖（provided 作用域）
→ 版本对齐 + provided 依赖避免冲突

- dbms.procedures() 确认注册（名字/签名）
- 调用冒烟测试（最小输入 → 期望输出）
- 性能冒烟（确认不拖垮查询）
- 白名单检查（allowlist 是否放行）
→ 部署验证：注册 + 冒烟 + 性能 + 白名单四查

- 团队过程库：独立 jar + 统一命名空间
- 版本：过程库版本与 Neo4j 版本对齐管理
- 文档：每个过程记录用途/参数/返回
- 发布：内部仓库（Nexus/Artifactory）分发
→ 过程库要像「内部包」一样版本化 + 文档化

- 批量操作 > 逐条循环（批处理）
- 索引优先：过程内访问节点用索引（findNode 需要索引）
- 避免 N+1：批查而非逐条查
- 事务批量提交：大写入分事务提交（periodic.iterate）
- 并发：parallel 参数 + 控制并发度
→ 过程性能 = 批量 + 索引 + 分批提交 + 受控并发

// 过程内批量：避免逐条 commit
try (Transaction tx = db.beginTx()) {
    for (int i = 0; i < 10000; i++) {
        tx.execute("CREATE (a:Account {id: $id})",
                    Map.of("id", "A" + i));
        if (i % 1000 == 0) { tx.commit(); tx = db.beginTx(); }
    }
    tx.commit();
}
// 分批提交：每 1000 条一个事务

1. 白名单：只放行可信过程（allowlist）
2. 权限：过程与角色绑定（GRANT EXECUTE PROCEDURE）
3. 输入校验：参数校验（防注入/恶意输入）
4. 资源限制：结果限制（防大返回拖垮客户端）
5. 外部调用谨慎：过程内调外部服务的鉴权/超时
→ 安全 = 白名单 + 权限 + 校验 + 资源限制

// 参数校验（防注入）
@Procedure("example.findUser")
public Stream<UserResult> findUser(@Name("id") String id) {
    if (id == null || id.isBlank()
        || id.length() > 64 || !id.matches("\\w+")) {
        throw new RuntimeException("Invalid id");
    }
    // 参数化查询（防 Cypher 注入）
    try (Transaction tx = db.beginTx()) {
        Node n = tx.findNode(Label.label("User"), "id", id);
        ...
    }
}

- 并发过程共享图 → 需注意锁（写冲突）
- 过程内并发：parallel 控制线程数（别全核开满）
- 事务边界：过程内事务与查询事务嵌套要清楚
- 资源：过程内限制结果集大小（LIMIT）
→ 并发 = 控制线程数 + 事务边界清晰 + 结果限流

- 对比 Cypher 等价实现（过程是否更快）
- 压力测试：并发调用过程
- 监控：过程调用频率/耗时/错误率
- 结果验证：过程输出与手工结果比对
→ 性能 = 基准对比 + 压测 + 监控 + 正确性验证

// 调用外部 HTTP 服务（如风控评分）
@Procedure("risk.score")
public Stream<ScoreResult> score(@Name("accountId") String id) {
    // 用 HttpClient 调外部服务
    double score = callExternalScoringApi(id);
    return Stream.of(new ScoreResult(id, score));
}
// Cypher 侧：查询时动态获得外部评分

- 场景：GDS 没有的特定算法（如时序衰减分数）
- 过程内实现迭代算法（Java 原生比 Cypher 循环快）
- 可访问图数据（遍历/聚合）
- 输出节点/分数行
→ 自定义算法 = 性能敏感时的 Java 实现

// 批量清洗/规范化属性
@Procedure("example.normalize")
public Stream<SummaryResult> normalize() {
    long updated = 0;
    try (Transaction tx = db.beginTx()) {
        ResourceIterator<Node> it =
            tx.findNodes(Label.label("Person"));
        while (it.hasNext()) {
            Node n = it.next();
            String name = (String) n.getProperty("name");
            if (name != null) {
                n.setProperty("name", name.trim());
                updated++;
            }
        }
        tx.commit();
        return Stream.of(new SummaryResult(updated));
    }
}

- 统一命名空间：com.example.domain.*
- 封装复杂业务逻辑（规则评估/打分/判定）
- 团队共享：内部 jar 发布
- 文档 + 版本管理（像内部 SDK）
→ 领域过程库 = 把「复杂业务逻辑」变成可调用资产

// 校验数据质量（返回违规行）
@Procedure("quality.check")
public Stream<ViolationResult> check() {
    // 扫描缺失属性/悬空关系
    // 返回违规详情（节点、类型、原因）
}
// 配合治理：定时调用 + 告警（见治理篇）

- NLP 结果入库：分词/实体识别过程
- 文档 → 实体 → 关系（结合 LLM/规则）
- 搜索增强：过程内计算相关性
→ 文本过程 = 非结构化 → 结构化图谱的桥