最大流¶

本文介绍最大流算法，并演示如何用 GQL 实现该算法。

算法介绍¶

最大流用于在容量约束下计算从源点到汇点可达到的最大流量。

该过程将算法逻辑封装为可复用过程，可直接在图数据上执行并输出结果。

更多信息，请参见最大流。

数据模型¶

为了在 GQL 中实现最大流算法，使用以下数据模型创建图类型：

实体由类型为Node的点表示，点属性包括id和rating。
实体间关系由类型为Link的边表示，边属性包括weight。

该图类型的完整定义，请参见准备图数据。

过程实现 ¶

为了计算最大流结果，创建一个过程。

关于该过程的更多信息，请参见过程实现。下文将介绍该过程的关键组成部分及其工作原理。

该过程接受以下输入参数：

source_id：INT64类型。流源点的 ID。
sink_id：INT64类型。汇点点的 ID。

此外，在过程内部定义了以下变量：

变量名	变量类型	描述
`frontier`	原始值变量	BFS 增广路径搜索的活跃点集合。
`has_path`	原始值变量	是否找到通往汇点的增广路径。
`inf`	原始值变量	无界容量的无穷大哨兵值。
`total_flow`	原始值变量	从源点到汇点的总累积流量。
`augment_count`	原始值变量	已完成的增广路径迭代次数。
`augment_cap`	原始值变量	当前增广路径的瓶颈容量。
`walk_id`	原始值变量	增广路径重建中使用的点 ID。
`prev_id`	原始值变量	重建时增广路径中的上一个点。
`edge_dir`	原始值变量	流量更新的方向标志（+1 正向，-1 反向）。
`bfs_active`	原始值变量	本步骤 BFS 是否至少扩展了一个新点。
`sink_found`	原始值变量	本轮 BFS 是否到达了汇点。
`sink_state`	全局 SumAgg 聚合值变量	汇点处记录的最优路径状态。
`back_state`	全局 SumAgg 聚合值变量	增广路径重建中使用的状态。
`out_cap`	点绑定 SumAgg 聚合值变量	每点的 {邻居 ID → 剩余容量} 映射。
`out_flow`	点绑定 SumAgg 聚合值变量	每点的 {邻居 ID → 当前流量} 映射。
`seen`	点绑定 SumAgg 聚合值变量	每点标志：该点在当前 BFS 轮次中已被访问。
`cand`	点绑定 SumAgg 聚合值变量	当前 BFS 扩展中该点的最优路径候选。
`state`	点绑定 SumAgg 聚合值变量	每点当前最优增广路径状态。

以下步骤概述了该过程的逻辑：

从边权初始化残差容量out_cap，将流量out_flow初始化为 0。
反复从源点执行 BFS，寻找到汇点的增广路径（sink_found），通过cand和state追踪路径状态。
从汇点回溯到源点，计算瓶颈容量（augment_cap），并沿路径更新out_cap和out_flow。
将augment_cap累加到total_flow；当无增广路径时停止。返回最大流总量。

操作步骤¶

在悦数图管理的控制台中，按顺序执行以下代码块。首次使用时，先按准备算法示例数据确认操作位置和目录挂载。

1. 准备图数据 ¶

创建一个与数据模型匹配的图类型。

'''
CREATE GRAPH TYPE IF NOT EXISTS maximum_flow_graph_type AS {
    NODE Node (LABEL Node {id INT PRIMARY KEY, rating INT}),
    EDGE Link (Node)-[LABEL Link {weight INT, MULTIEDGE KEY()}]->(Node)
}
'''

下载以下 CSV 文件，保留文件名，保存到部署目录下的examples/algorithms/maximum_flow/：

maximum_flow_node.csv：点数据。
maximum_flow_edge.csv：边数据。

查看 CSV 内容

maximum_flow_node.csv

id,rating
1,0
2,2
3,1
4,0
5,0
6,2
7,1
8,2
9,0
10,1
11,0
12,2
13,1
14,1
15,0
16,2
17,0
18,1
19,2
20,0
21,2
22,1
23,0
24,1

maximum_flow_edge.csv

src_id,dst_id,weight
1,2,4
1,3,2
1,4,7
1,5,9
2,3,1
2,4,6
2,5,5
2,6,2
3,4,3
3,5,4
3,6,8
4,6,2
4,7,6
4,8,5
5,6,4
5,8,7
5,9,6
6,8,1
6,9,5
6,10,6
7,9,2
7,10,9
8,9,3
8,11,6
8,12,4
9,10,4
9,12,7
9,13,2
10,12,1
10,13,8
10,14,4
11,12,2
11,14,5
11,15,7
12,14,3
12,15,6
12,16,5
13,15,2
13,16,7
14,15,4
14,17,8
14,18,2
15,16,1
15,18,5
15,19,3
16,18,3
16,19,6
17,18,2
17,20,7
17,21,5
18,19,4
18,21,9
18,22,6
19,21,1
19,22,5
20,21,3
20,23,6
20,24,8
21,22,2
21,23,4
21,24,7
22,24,4
23,24,1

基于图类型创建临时图，并将数据文件加载到临时图中。

CREATE TEMPORARY GRAPH IF NOT EXISTS #maximum_flow_graph TYPED maximum_flow_graph_type

'''
USE #maximum_flow_graph IMPORT INTO GRAPH {
      NODE (v@Node {id: id, rating: rating}) FROM DATAFILE {
          PATH: "file:///examples/algorithms/maximum_flow/maximum_flow_node.csv",
          FORMAT: "CSV",
          DELIMITER: ','
      },
      EDGE (id: src_id)-[e@Link {weight: weight}]->(id: dst_id) FROM DATAFILE {
          PATH: "file:///examples/algorithms/maximum_flow/maximum_flow_edge.csv",
          FORMAT: "CSV",
          DELIMITER: ','
      }
  }
'''

导入成功后，在控制台执行以下查询核对点数：

'''
USE #maximum_flow_graph
MATCH (n)
RETURN count(n) AS node_count
'''

node_count应为24。点数不符时，先按文件问题排查检查文件和导入结果。

2. 创建过程 ¶

创建实现最大流算法的过程。关于该过程的更多信息，请参见过程实现。

'''
CREATE OR REPLACE PROCEDURE maximum_flow(source_id INT64, sink_id INT64)
RETURNS (source_id INT64, sink_id INT64, max_flow DOUBLE, augmentation_count INT64)
COMMENT "Compute the maximum flow in a directed graph whose vertex primary key is id and whose edge capacity property is weight. Parallel directed edges are aggregated by (src_id, dst_id) before each residual search."
AS {
  VALUE frontier ACTIVE_SET

  VALUE has_path BOOLEAN = true
  VALUE inf DOUBLE = 1000000000000000000.0
  VALUE total_flow DOUBLE = 0.0
  VALUE augment_count INT64 = 0
  VALUE augment_cap DOUBLE = 0.0
  VALUE walk_id INT64 = 0
  VALUE prev_id INT64 = 0
  VALUE edge_dir INT64 = 0

  VALUE bfs_active OrAgg = false
  VALUE sink_found OrAgg = false
  VALUE sink_state TopKAgg<1, path_cap DOUBLE DESC, parent_id INT64 ASC, dir INT64 DESC>
  VALUE back_state TopKAgg<1, path_cap DOUBLE DESC, parent_id INT64 ASC, dir INT64 DESC>

  NODE VALUE out_cap MapAgg<INT64, SumAgg<DOUBLE>>
  NODE VALUE out_flow MapAgg<INT64, SumAgg<DOUBLE>>
  NODE VALUE seen OrAgg = false
  NODE VALUE cand TopKAgg<1, path_cap DOUBLE DESC, parent_id INT64 ASC, dir INT64 DESC>
  NODE VALUE state TopKAgg<1, path_cap DOUBLE DESC, parent_id INT64 ASC, dir INT64 DESC>

  MATCH (s)-[e]->(t)
  PER PATH {
    SET s.@out_cap += TUPLE(t.id, CAST(e.weight AS DOUBLE))
    SET s.@out_flow += TUPLE(t.id, 0.0)
  }

  IF source_id = sink_id THEN {
    SET has_path = false
  }

  WHILE has_path = true THEN {
    MATCH (n)
    PER NODE (n) {
      SET n.@seen = false
      SET n.@cand.clear()
      SET n.@state.clear()
    }

    MATCH (s)
    WHERE s.id = source_id
    PER NODE (s) {
      SET s.@seen = true
      SET s.@state += RECORD{path_cap: inf, parent_id: source_id, dir: 1}
    }
    FINALLY {
      SET frontier = s
    }

    SET @bfs_active = true
    SET @sink_found = false

    WHILE @bfs_active = true AND @sink_found = false THEN {
      SET @bfs_active = false

      MATCH (n)
      WHERE n.@seen = false
      PER NODE (n) {
        SET n.@cand.clear()
      }

      MATCH (u)-[e]->(v)
      WHERE u IN frontier
        AND v.@seen = false
        AND u.@out_cap.get(v.id) - u.@out_flow.get(v.id) > 0.0
      PER PATH {
        VALUE residual DOUBLE = 0.0
        VALUE candidate_cap DOUBLE = 0.0

        SET residual = u.@out_cap.get(v.id) - u.@out_flow.get(v.id)

        IF u.@state.min().path_cap < residual THEN {
          SET candidate_cap = u.@state.min().path_cap
        }
        ELSE {
          SET candidate_cap = residual
        }

        SET v.@cand += RECORD{path_cap: candidate_cap, parent_id: u.id, dir: 1}
      }

      MATCH (u)<-[e]-(p)
      WHERE u IN frontier
        AND p.@seen = false
        AND p.@out_flow.get(u.id) > 0.0
      PER PATH {
        VALUE residual DOUBLE = 0.0
        VALUE candidate_cap DOUBLE = 0.0

        SET residual = p.@out_flow.get(u.id)

        IF u.@state.min().path_cap < residual THEN {
          SET candidate_cap = u.@state.min().path_cap
        }
        ELSE {
          SET candidate_cap = residual
        }

        SET p.@cand += RECORD{path_cap: candidate_cap, parent_id: u.id, dir: -1}
      }

      MATCH (n)
      WHERE n.@seen = false AND n.@cand.size() > 0
      PER NODE (n) {
        SET n.@seen = true
        SET n.@state += n.@cand.min()
        SET @bfs_active += true

        IF n.id = sink_id THEN {
          SET @sink_found += true
        }
      }
      FINALLY {
        SET frontier = n
      }
    }

    IF @sink_found = false THEN {
      SET has_path = false
    }
    ELSE {
      SET @sink_state.clear()

      MATCH (n)
      WHERE n.id = sink_id
      PER NODE (n) {
        SET @sink_state += n.@state.min()
      }

      SET augment_cap = @sink_state.min().path_cap
      SET total_flow = total_flow + augment_cap
      SET augment_count = augment_count + 1
      SET walk_id = sink_id

      WHILE walk_id <> source_id THEN {
        SET @back_state.clear()

        MATCH (n)
        WHERE n.id = walk_id
        PER NODE (n) {
          SET @back_state += n.@state.min()
        }

        SET prev_id = @back_state.min().parent_id
        SET edge_dir = @back_state.min().dir

        IF edge_dir = 1 THEN {
          MATCH (p)
          WHERE p.id = prev_id
          PER NODE (p) {
            SET p.@out_flow += TUPLE(walk_id, augment_cap)
          }
        }
        ELSE {
          MATCH (p)
          WHERE p.id = walk_id
          PER NODE (p) {
            SET p.@out_flow += TUPLE(prev_id, -augment_cap)
          }
        }

        SET walk_id = prev_id
      }
    }
  }

  RETURN
    source_id AS source_id,
    sink_id AS sink_id,
    total_flow AS max_flow,
    augment_count AS augmentation_count
}
'''

3. 调用过程 ¶

以点1为源点、点24为汇点，计算最大流及增广次数。

USE #maximum_flow_graph CALL maximum_flow(1, 24) RETURN source_id, sink_id, max_flow, augmentation_count

返回：

+-----------+---------+----------+--------------------+
| source_id | sink_id | max_flow | augmentation_count |
+-----------+---------+----------+--------------------+
| 1         | 24      | 19.0     | 6                  |
+-----------+---------+----------+--------------------+

保存结果到文件或图中

如需将返回的结果保存为 CSV 文件或写入图，请参见 EXPORT。导出字段应与返回列及目标属性对应。

最大流¶

算法介绍¶

数据模型¶

过程实现¶

操作步骤¶

1. 准备图数据¶

2. 创建过程¶

3. 调用过程¶

过程实现 ¶

1. 准备图数据 ¶

2. 创建过程 ¶

3. 调用过程 ¶