基于MongoDB $graphLookup构建无向交易图谱的技术咨询

问题描述 #

需要用MongoDB的$graphLookup构建无向图处理人际交易数据，单条交易文档结构如下：

{
  "from": "A",
  "to": "B",
  "value": 1
}

要求：

• 节点为个人账号
• 边的volume代表两人间的交易总次数（例如A→B和B→A的两条交易需合并为一条volume为2的边）

当前疑问：

• 是否需要预处理原始交易数据再入库？
• 如何通过查询得到符合要求的无向图结果？

给定原始输入数据：

[
  {
    "from": "A",
    "to": "B",
    "value": 1
  },
  {
    "from": "B",
    "to": "A",
    "value": 3
  },
  {
    "from": "C",
    "to": "A",
    "value": 6
  },
  {
    "from": "C",
    "to": "A",
    "value": 10
  },
  {
    "from": "A",
    "to": "C",
    "value": 20
  }
]

期望查询结果（以A为起始节点）：

{
   "startedAddress": "A",
   "neighboors": [
      {"address": "C", "volume": 3, "depth": 1},
      {"address": "B", "volume": 2, "depth": 1}
   ]
}

解决方案 #

方案一：预处理数据后入库（性能更优） #

预处理的核心是统一每条交易的节点顺序，确保A-B和B-A被识别为同一组，再提前统计交易次数存入专门的边集合。

预处理步骤（生成标准化边集合）

db.transactions.aggregate([
  // 标准化节点顺序：node1为字典序较小的账号，node2为较大的账号
  {
    $addFields: {
      node1: { $min: ["$from", "$to"] },
      node2: { $max: ["$from", "$to"] }
    }
  },
  // 按标准化节点对分组，统计交易总次数
  {
    $group: {
      _id: { node1: "$node1", node2: "$node2" },
      volume: { $sum: 1 }
    }
  },
  // 将结果导出到新集合graph_edges
  { $out: "graph_edges" }
])

基于预处理集合的查询

利用$graphLookup实现无向遍历，通过$cond动态匹配节点的两个方向：

db.graph_edges.aggregate([
  // 筛选所有与A相关的边
  { $match: { $or: [{ "node1": "A" }, { "node2": "A" }] } },
  // 执行无向图遍历
  {
    $graphLookup: {
      from: "graph_edges",
      startWith: { $cond: [{ $eq: ["$node1", "A"] }, "$node2", "$node1"] },
      connectFromField: { $cond: [{ $eq: ["$_id.node1", "$$this.node1"] }, "$_id.node2", "$_id.node1"] },
      connectToField: { $cond: [{ $eq: ["$$this.node1", "$$this.node1"] }, "$$this.node1", "$$this.node2"] },
      as: "neighbors",
      depthField: "depth"
    }
  },
  // 整理为期望的输出格式
  {
    $project: {
      _id: 0,
      startedAddress: "A",
      neighboors: {
        $map: {
          input: "$neighbors",
          as: "n",
          in: {
            address: { $cond: [{ $eq: ["n.node1", "A"] }, "n.node2", "n.node1"] },
            volume: "n.volume",
            depth: "n.depth"
          }
        }
      }
    }
  }
])

方案二：直接查询时实时处理（无需修改原始数据） #

如果不想改动原始交易集合，可以在聚合查询中实时处理无向关系并统计交易次数：

db.transactions.aggregate([
  // 第一步：标准化节点顺序并统计交易次数
  {
    $addFields: {
      node1: { $min: ["$from", "$to"] },
      node2: { $max: ["$from", "$to"] }
    }
  },
  {
    $group: {
      _id: { node1: "$node1", node2: "$node2" },
      volume: { $sum: 1 }
    }
  },
  // 第二步：执行无向图遍历，内部管道实时处理邻居节点的交易统计
  {
    $graphLookup: {
      from: "transactions",
      startWith: "A",
      connectFromField: "address",
      connectToField: { $cond: [{ $eq: ["$_id.node1", "$$address"] }, "$_id.node2", "$_id.node1"] },
      as: "neighbors",
      depthField: "depth",
      pipeline: [
        {
          $addFields: {
            node1: { $min: ["$from", "$to"] },
            node2: { $max: ["$from", "$to"] }
          }
        },
        {
          $group: {
            _id: { node1: "$node1", node2: "$node2" },
            volume: { $sum: 1 }
          }
        },
        {
          $project: {
            address: { $cond: [{ $eq: ["$_id.node1", "$$address"] }, "$_id.node2", "$_id.node1"] },
            volume: 1,
            _id: 0
          }
        }
      ]
    }
  },
  // 第三步：筛选起始节点为A的结果并整理格式
  { $match: { $or: [{ "_id.node1": "A" }, { "_id.node2": "A" }] } },
  {
    $project: {
      _id: 0,
      startedAddress: "A",
      neighboors: {
        $map: {
          input: "$neighbors",
          as: "n",
          in: {
            address: "n.address",
            volume: "n.volume",
            depth: "n.depth"
          }
        }
      }
    }
  },
  // 去重邻居节点
  {
    $addFields: {
      neighboors: {
        $reduce: {
          input: "$neighboors",
          initialValue: [],
          in: {
            $cond: [
              { $in: ["$$this.address", "$$value.address"] },
              "$$value",
              { $concatArrays: ["$$value", ["$$this"]] }
            ]
          }
        }
      }
    }
  }
])

关键说明 #

• 无向关系的核心是统一节点对的表示逻辑，确保反向交易被识别为同一组，无论是预处理还是实时查询都要遵循这一原则。
• $graphLookup中通过$cond判断当前节点在标准化对中的位置，动态匹配关联节点，实现无向遍历。
• 预处理方案更适合需要频繁查询或深度遍历的场景，避免重复计算交易统计；实时处理方案则适合数据更新频繁、不想维护额外集合的场景。

内容的提问来源于stack exchange，提问作者nimrod feldman