PostgreSQL按可互换的源目IP端口对分组统计双向流量

输入NetFlow表数据 #

期望输出结果 #

最优实现思路（通用SQL方案） #

核心逻辑是为双向会话生成统一的标识键，确保正向、反向的流量记录被归为同一组，再通过聚合计算拆分双向字节数。以下是兼容绝大多数关系型数据库的实现代码：

WITH flow_with_session AS (
    SELECT
        src_ip,
        src_port,
        dst_ip,
        dst_port,
        bytes_sent,
        -- 生成统一会话键：按IP+端口的字典序排序，确保双向记录键一致
        CASE
            WHEN CONCAT(src_ip, ':', src_port) <= CONCAT(dst_ip, ':', dst_port)
            THEN CONCAT(src_ip, ':', src_port, '|', dst_ip, ':', dst_port)
            ELSE CONCAT(dst_ip, ':', dst_port, '|', src_ip, ':', src_port)
        END AS session_key,
        -- 标记流量方向，用于后续区分发送/接收字节
        CASE
            WHEN CONCAT(src_ip, ':', src_port) <= CONCAT(dst_ip, ':', dst_port)
            THEN 'original'
            ELSE 'reverse'
        END AS direction
    FROM netflow_table
),
session_agg AS (
    SELECT
        session_key,
        -- 保留原始方向的IP/端口作为结果的src/dst字段
        MAX(CASE WHEN direction = 'original' THEN src_ip END) AS src_ip,
        MAX(CASE WHEN direction = 'original' THEN src_port END) AS src_port,
        MAX(CASE WHEN direction = 'original' THEN dst_ip END) AS dst_ip,
        MAX(CASE WHEN direction = 'original' THEN dst_port END) AS dst_port,
        -- 汇总原始方向的总发送字节
        SUM(CASE WHEN direction = 'original' THEN bytes_sent ELSE 0 END) AS bytes_sent,
        -- 汇总反向方向的总接收字节（无反向则为0）
        SUM(CASE WHEN direction = 'reverse' THEN bytes_sent ELSE 0 END) AS bytes_recv
    FROM flow_with_session
    GROUP BY session_key
)
SELECT src_ip, src_port, dst_ip, dst_port, bytes_sent, bytes_recv FROM session_agg;

性能优化建议 #

如果处理海量NetFlow数据，可替换字符串拼接的会话键为复合行键（部分数据库如PostgreSQL支持），避免字符串操作的性能损耗：

-- PostgreSQL 优化版会话键生成
LEAST(ROW(src_ip, src_port), ROW(dst_ip, dst_port)) AS session_part1,
GREATEST(ROW(src_ip, src_port), ROW(dst_ip, dst_port)) AS session_part2

后续分组时直接用session_part1和session_part2作为分组键即可，性能更优。

如果是分布式场景（如Spark SQL处理NetFlow数据），逻辑完全一致，仅需调整语法适配对应引擎。

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