如何在Gremlin-Java中使用addV动态Upsert未知数量顶点？

一、动态构建顶点Upsert遍历的Java实现 #

针对小批量（1-100条）顶点的Upsert需求，核心思路是利用Gremlin的fold()+coalesce()组合实现“存在则更新，不存在则插入”的逻辑，再通过Java链式调用动态拼接属性，规避泛型适配问题。

单条顶点Upsert（逐次提交） #

import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversal;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__;
import org.apache.tinkerpop.gremlin.structure.Vertex;
import java.util.Map;

public class VertexUpsertUtil {

    private void upsertSingleVertex(GraphTraversalSource g, Map<String, Object> vertexData) {
        String label = (String) vertexData.get("label");
        String uniqueKey = (String) vertexData.get("uniqueKey");
        Object uniqueValue = vertexData.get(uniqueKey);

        // 基础Upsert逻辑：先查询目标顶点，不存在则创建
        GraphTraversal<?, Vertex> traversal = g.V().has(uniqueKey, uniqueValue)
                .fold()
                .coalesce(
                        __.unfold(), // 存在则取出已有顶点
                        __.addV(label).property(uniqueKey, uniqueValue) // 不存在则新建顶点
                );

        // 动态添加剩余属性
        for (Map.Entry<String, Object> entry : vertexData.entrySet()) {
            String key = entry.getKey();
            if (!key.equals("label") && !key.equals(uniqueKey)) {
                traversal.property(key, entry.getValue());
            }
        }

        traversal.next();
    }

    // 批量处理顶点
    public void batchUpsert(GraphTraversalSource g, Iterable<Map<String, Object>> vertexDataList) {
        for (Map<String, Object> data : vertexDataList) {
            upsertSingleVertex(g, data);
        }
    }
}

批量顶点Upsert（单次请求提交） #

如果想减少网络请求开销，可将所有Upsert逻辑通过union()合并为一个遍历，一次性提交到服务器：

// 批量Upsert合并为单次请求
public void batchUpsertSingleRequest(GraphTraversalSource g, Iterable<Map<String, Object>> vertexDataList) {
    GraphTraversal<?, ?> batchTraversal = g.union();

    for (Map<String, Object> data : vertexDataList) {
        String label = (String) data.get("label");
        String uniqueKey = (String) data.get("uniqueKey");
        Object uniqueValue = data.get(uniqueKey);

        // 为每个顶点构建独立的Upsert分支
        GraphTraversal<?, Vertex> vertexTraversal = __.V().has(uniqueKey, uniqueValue)
                .fold()
                .coalesce(
                        __.unfold(),
                        __.addV(label).property(uniqueKey, uniqueValue)
                );

        // 动态追加属性
        for (Map.Entry<String, Object> entry : data.entrySet()) {
            String key = entry.getKey();
            if (!key.equals("label") && !key.equals(uniqueKey)) {
                vertexTraversal.property(key, entry.getValue());
            }
        }

        // 将分支加入批量遍历
        batchTraversal = batchTraversal.add(vertexTraversal);
    }

    // 执行批量操作
    batchTraversal.iterate();
}

说明：

• 无需使用Function#andThen，Gremlin的链式调用本身就是动态构建遍历的最优方式，泛型会自动衔接每一步的GraphTraversal类型。
• fold()将查询结果转为集合，coalesce()优先处理已有顶点、否则执行新建逻辑，这是TinkerPop官方推荐的跨后端Upsert实现。

二、后端无关的顶点加载工具 #

1. Gremlin Console脚本加载 #

利用Gremlin Console（Groovy脚本）读取CSV并批量Upsert，支持本地TinkerGraph和远程Gremlin Server（包括Neptune）：

// 连接图实例：本地TinkerGraph用 g = TinkerGraph.open().traversal()
// 远程Neptune用 g = traversal().withRemote('conf/neptune-remote.properties')

// 读取CSV文件（示例格式：label,uniqueKey,uniqueVal,prop1,prop2）
new File('vertices.csv').eachLine { line ->
    def cols = line.split(',')
    def label = cols[0]
    def uniqueKey = cols[1]
    def uniqueVal = cols[2]

    def traversal = g.V().has(uniqueKey, uniqueVal).fold()
        .coalesce(unfold(), addV(label).property(uniqueKey, uniqueVal))
    
    // 动态添加属性（从第4列开始，键值对交替）
    for (int i = 3; i < cols.size(); i += 2) {
        if (i+1 < cols.size()) {
            traversal.property(cols[i], cols[i+1])
        }
    }
    
    traversal.iterate()
}

2. 自定义Java工具类 #

前面实现的VertexUpsertUtil就是完全后端无关的工具，只需切换GraphTraversalSource的连接方式：

• 本地TinkerGraph：GraphTraversalSource g = TinkerGraph.open().traversal();
• 远程Gremlin Server/Neptune：GraphTraversalSource g = traversal().withRemote(DriverRemoteConnection.using("host:port"));

3. Apache NiFi（数据流场景） #

如果是持续数据流加载，可使用NiFi的GremlinQuery或GremlinBulkLoad处理器，配置Upsert逻辑后，从CSV、Kafka等数据源读取数据，自动批量写入图数据库，无需关心具体后端类型。

内容的提问来源于stack exchange，提问作者Fabrice Gabolde