NetFlow 数据清洗并不是简单删掉几行坏记录，而是一个需要从字段特征、阈值策略和统计模型多个层面考虑的过程。原始 NetFlow 导出数据里经常混入设备重启时产生的零字节记录、采样误差导致的超大流量，以及扫描攻击留下的短命 TCP 流。它们如果不加处理直接进入分析系统，会拉高带宽基线、触发误报，甚至掩盖真实攻击。下面这段 Ruby 代码只是全文的一个引子，实际清洗逻辑还要结合业务网络的规模做调整。
本文以 CSV 格式的 NetFlow 导出数据为例，先说明字段特征，再给出一套可以运行的 Ruby 清洗实现，最后讨论如何用统计方法替代写死的阈值。你可以根据实际采集器输出调整字段名和阈值。
NetFlow记录中哪些字段容易产生异常值
NetFlow 的每条记录通常包含源目的地址、端口、协议、字节数、包数、起始时间和持续时间等字段。清洗异常值时，最常用的判断依据集中在字节数、包数和持续时间上。一条正常的 HTTP 会话可能有几 KB 到几 MB 的字节数，持续时间从几百毫秒到几分钟不等。而异常记录往往表现为字节数极大但包数很小，或者包数极大但持续时间为零，这两种组合都可能说明设备在聚合时出现了错误。
TCP 标志位也值得单独关注。正常的 TCP 连接会经历三次握手，SYN 和 ACK 会成对出现。如果一条流的标志位只有 RST 或 FIN，同时没有 SYN，它很可能是一次未完成的扫描或伪造流量。在过滤逻辑里把这类记录单独标记出来，可以显著减少误报。不同网络环境的基线不一样，比如数据中心内部的东西向流量可能天然包含大量短连接，所以阈值不能照搬，需要结合历史数据做分位数分析。
用Ruby实现一个基础清洗流水线
先把清洗逻辑落成代码。下面这个 NetFlowCleaner 类从 CSV 文件读取 NetFlow 导出数据，按字段做基础过滤。CSV 是很多 NetFlow 采集器支持的导出格式，如果用二进制 NetFlow v9 或 IPFIX，可以先用 nfcapd 或 Ruby 的 ipfix 库转成文本行。代码里设定字节数、包数和持续时间阈值，并调用 suspicious_tcp_flags? 检查 TCP 标志位。
require 'csv'
class NetFlowCleaner
BYTE_THRESHOLD = 10_000_000
PACKET_THRESHOLD = 1_000_000
FLOW_DURATION_MIN = 0.01
def initialize(file_path)
@file_path = file_path
end
def clean
cleaned = []
CSV.foreach(@file_path, headers: true) do |row|
bytes = row['bytes'].to_i
packets = row['packets'].to_i
duration = row['duration'].to_f
flags = row['tcp_flags'].to_i
next if bytes.zero? and packets.zero?
next if bytes > BYTE_THRESHOLD and packets > PACKET_THRESHOLD
next if duration < FLOW_DURATION_MIN
next if suspicious_tcp_flags?(flags)
cleaned << row.to_h
end
cleaned
end
def suspicious_tcp_flags?(flags)
syn_flag = 0x02
rst_flag = 0x04
fin_flag = 0x01
has_rst_or_fin = (flags & (rst_flag | fin_flag)) != 0
has_syn = (flags & syn_flag) != 0
has_rst_or_fin and not has_syn
end
end
cleaner = NetFlowCleaner.new('netflow.csv')
clean_records = cleaner.clean
puts "清洗后保留 #{clean_records.size} 条记录"
这段代码的思路是用 CSV.foreach 逐行读取，避免一次性把整个文件载入内存。对每条记录先解析出关键字段，再用 next if 跳过不符合条件的行。零字节且零包的记录通常没有分析价值，直接丢弃。字节数和包数同时超过阈值时，可能是设备聚合异常或异常大流量，也需要剔除。持续时间小于 10 毫秒的流往往只有一两个包，如果不关心扫描行为，可以过滤掉。
不过这段代码有一个明显局限：阈值是写死的。对于流量波动较大的网络，固定阈值可能会把正常的突发大流量误判为异常，也可能漏掉某些慢速攻击。因此在实际项目中，更推荐把阈值做成配置项，并定期根据历史数据重新计算。下面的统计方法可以解决一部分问题。
从固定阈值升级到稳健统计检测
NetFlow 的字节数和包数分布通常不是正态分布，而是典型的右偏长尾分布。大部分流都很小，少数流极大，平均值会被这些大值带偏。如果用平均值加标准差来判断异常，很多正常的大流量会被误杀。这时更适合使用中位数和绝对中位差，也就是 MAD。MAD 对离群值的敏感度远低于标准差，即使样本里混入少量极端值，也不会让判断基准大幅偏移。
下面这段代码演示如何计算给定样本的 MAD，并根据稳健 Z 分数判断某个值是否为异常。阈值 3.5 是统计上常用的保守值，可以按告警容忍度调整。对于实时 NetFlow 清洗，可以维护一个滑动窗口，例如每 5 分钟计算一次窗口内的分位数，再把当前记录与窗口基准比较。
def median(values)
sorted = values.sort
mid = sorted.length / 2
sorted.length.even? ? (sorted[mid - 1] + sorted[mid]) / 2.0 : sorted[mid]
end
def mad(values)
med = median(values)
deviations = values.map { |v| (v - med).abs }
median(deviations)
end
def outlier?(value, values, threshold = 3.5)
med = median(values)
mad_value = mad(values)
return false if mad_value.zero?
score = 0.6745 * (value - med).abs / mad_value
score > threshold
end
bytes_sample = [1200, 1340, 880, 19900, 1500, 1400, 2100, 1600, 1300]
puts outlier?(19900, bytes_sample, 3.5)
代码先对样本排序求中位数，再计算每个值与中位数的绝对偏差，最后取这些偏差的中位数得到 MAD。异常判断时用常数 0.6745 做标准化，让 MAD 在正态分布下近似标准差。返回 false 的情况包含 MAD 为零，这意味着样本高度一致，不应该触发异常。这个实现适合在 NetFlow 清洗脚本里单独封装成一个模块，避免与业务逻辑耦合。
需要注意的是，MAD 方法也有代价。它需要对滑动窗口内的数据排序，窗口越大计算越慢。如果 NetFlow 记录量达到每秒数万条，全量排序会拖垮清洗性能。此时可以改用近似分位数算法，比如 Greenwald-Khanna 算法或直接使用 Apache DataSketches 的 Ruby 绑定，牺牲一点精度换取吞吐量。
生产环境落地的几个关键点
把上面的清洗逻辑放进生产环境，首先要解决数据接入问题。NetFlow 通常以 UDP 包形式从路由器发到采集器，采集器再导出成文件或消息流。直接用 Ruby 监听 UDP 端口并解析 NetFlow v9 模板虽然可行，但容易因为模板丢失导致解析中断。更稳妥的做法是让 nfcapd 或 pmacct 负责采集和预处理，Ruby 只消费它们生成的 CSV 或 JSON 行，专注于清洗和后续分析。
另一个容易忽略的问题是内存控制。cleaned << row.to_h 会把所有清洗后的记录保存在数组里，数据量一大就会撑爆内存。生产代码应该把清洗后的记录直接写入下游，比如追加到文件、发送到 Kafka 或插入 ClickHouse。如果确实需要批量操作，可以每积累 5000 条 flush 一次，避免数组无限增长。同时要记录过滤原因，例如用计数器统计因字节阈值、持续时间或 TCP 标志被丢弃的数量，方便回溯误杀。
最后是异常检测的反馈闭环。清洗规则上线后，需要定期抽查被过滤的记录，看看是否存在正常业务被误删。可以把被过滤的高风险记录输出到单独的审计日志，让安全或网络团队复查。只有把规则效果和实际网络行为对齐，NetFlow 异常值清洗才不会变成另一个盲目丢弃数据的黑盒。
