假设你手上有10台业务服务器每台都在不停吐日志Flume的单机采集玩法你已经跑得很顺了。可一旦要把这10台机器的日志统一汇到一个地方事情就没那么简单HDFS上出现一堆彼此独立的目录业务方取数前得自己做合并小文件的数量也在肉眼可见地增加。这篇是Flume快速入门系列的第6篇主题就是聚合也就是让多个采集端Agent先把日志推送到一个汇总端Agent再由汇总端Agent统一负责落盘。我会把聚合的拓扑原理、Avro Source与Avro Sink两端的配置配合、一套可复制改IP的双Agent Demo以及聚合链路里最常见的几个坑一次讲清楚。1. 聚合到底解决什么问题从单机直写HDFS到多机汇流1.1 单Agent直写HDFS的三个痛点很多团队一开始的做法很朴素每台机器上一个Flume Agenttaildir或spooldir读本地日志hdfs sink直接往HDFS写。机器少的时候看不出来问题一旦节点上了两位数三个痛点会非常扎眼。第一是目录分裂。每个Agent都会按照hdfs.path里的时间戳规则自己建目录于是HDFS上出现N个并行的日志目录。业务方要做离线分析得先写一段脚本把这N个目录拉平合并数据口径稍微不一致排查就是一场灾难。第二是小文件激增。每个Agent的HDFS sink都在独立控制文件滚动默认rollInterval只有30秒10台机器同时滚动一小时内就能产生上千个几十KB的小文件。NameNode维护元数据的压力陡增后续Spark、Hive扫描这些小文件时性能也会明显下降。第三是运维成本失控。改了采集规则要一台一台机器去更新配置想统计全局采集吞吐得逐台登录看日志。你不是在管理一套数据管道而是在管理10套互不相干的管道。1.2 聚合的扇入模型采集端与汇总端各干各的活聚合解决的就是上面这三个问题。Flume官网把这种拓扑叫扇入也就是多个Agent的数据流汇入同一个Agent。它的结构非常清晰最上游是一堆采集端Agent每个Agent负责一台机器用taildir之类的Source读本地日志然后通过Avro Sink把Event发到网络下游是一个汇总端Agent它用Avro Source接住所有采集端发来的数据再统一交给HDFS Sink这种真正的落盘Sink。这里要理解一个关键点汇总端Agent并没有特殊身份它本身也是一个标准的Source-Channel-Sink三段结构只不过它的Source是专门为接收跨网络Event而生的Avro Source。整条链路每一跳都是独立的事务我在后面第4节会展开讲。这种分工带来的直接好处就是目录统一由汇总端控制所有日志都落在同一个根路径下文件滚动只发生在汇总端一个节点小文件数量大幅收敛运维也只需要重点盯着汇总端这一个节点采集端的配置改动变成了下发文件的事。1.3 跨Agent通信为什么默认选AvroFlume家族里有好几种Source和Sink组合可以跨网络传数据但Agent与Agent之间的聚合通信官方和社区默认都是Avro原因有三。Avro Source和Avro Sink是Flume原生对原生Event在传输过程中能完整保留headers。别小看headers这是后续做目录分流、下游动态过滤的基础。比如我在第3节的Demo里会用static拦截器给每个Event打上host标记汇总端就能按host自动分目录这个能力只有原生传输才能无损带过去。第二个原因是Avro两端配合成熟稳定有独立的连接超时、请求超时、批次大小参数还支持deflate或snappy压缩。相比之下Syslog Source是为了兼容标准syslog协议Netcat Source基本只配测试环境使用Thrift则是给Thrift生态的客户端留的接口都不如Avro适合Agent之间高速、稳定的数据搬运。2. Avro两端参数详解Sink负责推送Source负责接住2.1 采集端Avro Sink配置把Event发给谁Avro Sink在采集端扮演的是客户端角色它从Channel里取Event然后通过网络发送到汇总端。核心参数不多但每一个都值得认真对待。参数默认值说明typeavro固定写法hostname无汇总端Agent所在机器的IP或域名port无汇总端Avro Source监听的端口两端必须一致batch-size100每个批次尝试发送的Event条数建议与Channel的transactionCapacity联动connect-timeout20000建立连接的超时时间单位毫秒request-timeout20000发送一次请求的超时时间单位毫秒compression-typenull可选null、deflate、snappy数据量大时开启压缩能省带宽hostname和port是最容易埋坑的两个参数。我在排查聚合故障时发现相当一部分采集端报错连不上的问题都是port写岔了或者汇总端改过端口忘了同步给下游。这里建议把端口号纳入统一的配置管理而不是散落在每台机器的agent配置里。2.2 汇总端Avro Source配置怎么接住多路数据Avro Source在汇总端扮演的是服务端角色负责监听端口、接收多个采集端Agent推过来的数据。参数更少但有个bind字段值得单独解释。参数默认值说明typeavro固定写法bind无绑定哪个网卡地址生产环境建议0.0.0.0port无监听端口必须与上游Avro Sink的port一致threads10处理网络请求的工作线程数上游Agent较多时需要调大bind这个字段我在生产环境一律写0.0.0.0表示监听所有网卡接口。有人喜欢写具体IP这在单网卡机器上没问题但一旦机器有多块网卡、或者网络拓扑做了调整很容易出现明明服务在跑但上游就是连不上的诡异故障。用0.0.0.0配合防火墙白名单是更稳妥的方案。threads的含义是有多少个线程并行处理网络请求。它不是一个约等于连接数的参数而是并行工作线程数。上游Agent数量少时用默认10没问题上游超过二三十个时建议按采集端数量的一定比例调大否则会出现一个线程排队处理多个连接的情况拖慢整体吞吐。2.3 两端笔数约束batch-size与transactionCapacity的数量关系这是聚合配置里最容易被忽视的数学关系。一条Event从采集端的磁盘文件到汇总端的HDFS要经历四段交易采集端Source从文件读Event放入Channel采集端Sink从Channel取Event发到网络汇总端Source从网络接收Event放入自己的Channel汇总端Sink从Channel取Event写入HDFS。每一段都是独立事务事务容量由Channel的transactionCapacity决定。当Avro Sink的batch-size设为200时意味着它一次事务要从Channel里取200条Event汇总端的Avro Source收到这批Event后也要一次性放入自己的Channel因此汇总端Channel的transactionCapacity必须大于等于上游端的batch-size。否则就会出现上游正常发送、下游Channel事务放不下的情况日志里全是重试报错数据吞吐断崖式下跌。在实际配置里我会让采集端Avro Sink的batch-size与本地Channel的transactionCapacity保持一致或留余量同时要求汇总端Channel的transactionCapacity大于所有上游batch-size的最大值。这套约束想清楚了聚合链路才能跑得顺。3. 从零搭一套聚合Demo双采集端汇聚到汇总端3.1 环境规划与安装前提演示环境我用三台机器全部安装Flume 1.9.0JDK 1.8及以上即可。三台机器不需要搭什么集群Flume Agent本来就是独立进程只要网络互通就行。机器角色示例IP在链路里的职责采集端 web01192.168.1.21taildir读日志Avro Sink发给汇总端采集端 web02192.168.1.22taildir读日志Avro Sink发给汇总端汇总端 agg01192.168.1.30Avro Source接收数据HDFS Sink落盘如果你只有一台机器也能跑通整个Demo在两个不同目录里启动两个Flume进程用localhost互相连接但端口必须错开agent名字也不能重复。单机演练对于理解链路很有帮助我建议第一次动手的读者先这么试。3.2 采集端配置taildir Source加static拦截器打标记web01的配置如下web02只需要把拦截器里的value改成web02即可。# 文件/opt/flume/conf/web01.conf web01.sources s1 web01.channels ch1 web01.sinks k1 # taildir逐行跟踪追加日志 web01.sources.s1.type taildir web01.sources.s1.positionFile /opt/flume/data/taildir-position.json web01.sources.s1.filegroups f1 web01.sources.s1.filegroups.f1 /opt/logs/app\\.log web01.sources.s1.interceptors i1 web01.sources.s1.interceptors.i1.type static web01.sources.s1.interceptors.i1.key host web01.sources.s1.interceptors.i1.value web01 web01.sources.s1.channels ch1 # Channel内存通道 web01.channels.ch1.type memory web01.channels.ch1.capacity 20000 web01.channels.ch1.transactionCapacity 1000 # Avro Sink数据发往汇总端 web01.sinks.k1.type avro web01.sinks.k1.hostname 192.168.1.30 web01.sinks.k1.port 4141 web01.sinks.k1.batch-size 200 web01.sinks.k1.channels ch1这里有两个细节。第一是taildir的filegroups正则properties文件里\\会被解析成一个反斜杠最终传给正则的是\.匹配真正的点号。如果直接写单反斜杠反而会被Properties吃成普通点号正则意义就变了。第二是static拦截器。它给每条Event的headers里塞了一个键值对web01标记为hostweb01web02标记为hostweb02。这个标记会在Avro传输中完整带到汇总端汇总端就能利用它做目录分流。这也是聚合场景里最有价值的技巧之一。3.3 汇总端配置Avro Source接货HDFS Sink统一落盘# 文件/opt/flume/conf/agg01.conf agg01.sources avroSrc agg01.channels ch agg01.sinks hdfsSink # Avro Source监听所有网卡等待上游连接 agg01.sources.avroSrc.type avro agg01.sources.avroSrc.bind 0.0.0.0 agg01.sources.avroSrc.port 4141 agg01.sources.avroSrc.threads 8 agg01.sources.avroSrc.channels ch # Channel汇总端建议给大容量 agg01.channels.ch.type memory agg01.channels.ch.capacity 50000 agg01.channels.ch.transactionCapacity 1000 # HDFS Sink按host和日期分目录 agg01.sinks.hdfsSink.type hdfs agg01.sinks.hdfsSink.hdfs.path /flume/logs/%{host}/%y-%m-%d agg01.sinks.hdfsSink.hdfs.filePrefix events agg01.sinks.hdfsSink.hdfs.fileType DataStream agg01.sinks.hdfsSink.hdfs.writeFormat Text agg01.sinks.hdfsSink.hdfs.rollInterval 300 agg01.sinks.hdfsSink.hdfs.rollSize 0 agg01.sinks.hdfsSink.hdfs.rollCount 0 agg01.sinks.hdfsSink.hdfs.batchSize 100 agg01.sinks.hdfsSink.channels chhdfs.path里的%{host}会从Event的headers中取host这个字段的值。如果上游没有这个header路径就会拼接异常表现可能是路径缺了一段也可能直接导致落盘报错。所以生产环境使用header做目录分流前一定要先确认上游所有Event都打了这个标记。我自己的习惯是先用logger sink或者file_roll sink验证链路完整再切到HDFS Sink。如果你没有HDFS环境把sink换成file_roll就可以快速验证聚合链路agg01.sinks.hdfsSink.type file_roll agg01.sinks.hdfsSink.sink.directory /opt/flume/output注意agent配置的agent名必须与启动命令里的-n参数一致配置文件里的前缀就是要和这个agent名对齐。我在web01.conf里写的是web01那么启动命令也必须用-n web01。3.4 启动顺序与验证先起汇总端再起采集端启动顺序是个很实际的经验。先启动汇总端agg01等Avro Source已经监听4141端口再逐台启动采集端。如果反着来采集端Agent一启动就会尝试连接汇总端连不上就会反复重试并打印WARN日志虽然不会丢数据但会污染整个启动过程的问题排查。启动命令如下三台机器各自执行# 汇总端 agg01 上执行 flume-ng agent -n agg01 -c conf -f conf/agg01.conf -Dflume.root.loggerINFO,console # 采集端 web01 上执行 flume-ng agent -n web01 -c conf -f conf/web01.conf -Dflume.root.loggerINFO,console验证分三步走。第一步在web01上往日志文件追加数据echo {user:alice,action:login} /opt/logs/app.log第二步回到agg01的终端看Flume日志如果链路正常看到类似EventDrainSuccessCount这样的计数器在增长就说明Event已经进入汇总端并成功落盘。第三步确认最终文件内容hdfs dfs -ls /flume/logs/web01/2025-01-05/ hdfs dfs -cat /flume/logs/web01/2025-01-05/events*能看到刚才追加的那条JSON就说明整条聚合链路完全打通了。web02上的流程完全一致最终数据会自动归到/web02对应的目录下。4. 聚合链路调优事务边界、批次大小与Channel蓄水4.1 端到端事务视角每一跳都是独立承诺聚合链路最容易让人困惑的地方在于它不是一条一次到位的管道而是由多个独立事务拼成的接力赛。所谓独立事务是指每一跳的Source成功把Event放进Channel、Sink成功把Event送出去并从Channel移除之后这一跳才真正确认成功。举个例子。web01的Avro Sink从ch1里取200条Event发往agg01agg01的Avro Source收到这200条并放入自己的Channel紧接着agg01的HDFS Sink从Channel里取数据写文件。如果agg01在收到数据后、文件写入前突然重启这200条Event并不会丢它们还在agg01的Channel里重启后会被继续消费。这就是人为理解的at-least-once语义聚合链路上的事件至少会被投递一次可能会重复但正常流程下不会丢失前提是Channel不溢出、也没有人为清空。理解这一点对定位聚合故障很有帮助——看到某个计数器涨了但文件里没有多数情况下不是丢数据而是事务还没走完。4.2 batch-size的选择逻辑吞吐与延迟的权衡batch-size太小比如10延迟低但同样的数据量需要更多次网络交互和事务提交CPU和网络开销都上去了吞吐上不来。batch-size太大比如2000单次事务的吞吐确实高但事务失败时的重试代价也大而且对两端Channel的transactionCapacity要求水涨船高。我的建议是从默认的100起步日志量大且链路稳定时调到200到500。调大batch-size的同时必须同步检查两件事一是采集端和汇总端的Channel transactionCapacity是否大于等于这个batch-size二是汇总端Avro Source的threads是否能支撑并发。这三者是联动的只调一个数值而没有配套调整很容易从正常变成间歇性卡顿。4.3 Channel的蓄水池效应为什么汇总端要特别大的容量聚合链路上汇总端是整个扇入模型的咽喉。所有采集端的流量都汇聚到这里它的Channel就是天然的蓄水池。上游瞬时流量增大时汇总端Source来不及消费的Event会先淤积在Channel里等流过去了再慢慢消化。因为这个角色汇总端的Channel容量我会明显给大memory类型的capacity给到50000到100000都是常见做法。但容量大不等于可靠memory Channel的数据活在内存里进程一旦重启没消费的Event全部丢失。对于日志分析管线通常可以接受如果是计费、风控这类不容错的场景汇总端就应该改用file Channel把Event先落到磁盘再消费代价是吞吐会有一定下降。核心思路是采集端可以用memory换吞吐汇总端这种单点汇聚环节必须优先保可靠。4.4 从两机汇聚扩展到多级聚合Demo里只有两个采集端连一个汇总端真实环境往往是几十甚至上百台机器。这时候把所有采集端直接接到一个汇总端上单点风险和网络连接压力都会上来。更常见的做法是两级聚合每十台左右的机器先汇聚到一个区域Agent多个区域Agent再汇到最终的总汇总端。每一级汇聚本质上都是同构的无非是上一级的Avro Sink指向下一级的Avro Source。但要注意两点一是链路变长后端到端延迟会叠加不适合对时效要求极高的场景二是每一级都需要独立监控和容量规划否则某一级Channel溢出故障会像多米诺骨牌一样向后传递。前面提到的batch-size和transactionCapacity约束在多级链路里每一跳都要满足。5. 聚合场景踩坑实录从端口不通到数据重复5.1 汇总端连不上的完整排查链路聚合场景里最高频的问题就是上游Agent报错连不上汇总端日志里反复出现类似无法连接、正在重试的信息。遇到这个我建议按下面的顺序排查而不是直接改配置重启。先确认汇总端的Avro Source确实在监听。在agg01上执行netstat -an | grep 4141看有没有LISTEN状态的记录。没有说明Flume进程没起来或者Source配置没生效先回头检查agg01.conf和启动日志。确认端口监听了再到采集端web01上执行telnet 192.168.1.30 4141测一下网络连通性。连不上基本就是防火墙、安全组或者物理网络的问题跟Flume配置无关了。一个很隐蔽的坑有些机器上Flume监听的端口被本机其他进程占用导致Source绑定失败这类问题在agg01的启动日志里会有明确的Address already in use。如果telnet通但还是报错检查Avro Sink的hostname和port与汇总端配置是否完全一致。我碰到过最典型的情况是汇总端从端口4141改成4142但采集端配置文件漏改了出现一批Agent间歇性连不上但又不完全报错的情况。这类问题靠日志里的目标地址一对比就能发现。5.2 数据少了一条交易边界与flush时机带来的错觉另一个高频困惑是采集端的计数明明增加了汇总端也增加了文件里却找不到最新那条数据。很多初学者会怀疑丢数据其实大概率是Flush延迟在作祟。HDFS Sink并不是每收到一条Event就立刻物理写入文件的它会按hdfs.batchSize攒批默认100条才flush一次。再加上文件滚动策略数据可能已经在内存里、在文件写缓冲里但还没真正落到磁盘。这个阶段你去看文件内容自然看不到最新的一条。Flume日志里计数器已经加了一但它统计的是Sink成功提交的事务数不等于文件系统可见的字节数。遇到这种情形不要慌。先看文件的修改时间和大小是否有变化再等一个滚动周期重新查看。真正的数据丢失往往伴随的是重试报错或者Channel溢出警告光凭计数差异判断会误伤自己。5.3 数据重复无法回避at-least-once的代价聚合链路里另一个绕不开的话题是重复。只要链路里存在网络中断、进程重启某些Event就可能被重复投递。原因在于事务的确认时点上游Sink发送成功就从Channel移除但下游可能还没来得及写入持久层就崩溃了重启后数据还在Channel里于是被再消费一次。应对方式取决于下游用途。离线日志分析通常对少量重复不敏感可以忽略。如果下游是统计去重、精确计数这类场景就必须在业务侧做幂等基于系统时间、日志序号或业务ID做去重而不是指望Flume本身给exactly-once。这一点在架构聚合链路时就该有个明确预期免得上了生产才发现语义对不上。5.4 务必验证headers再切换到HDFS最后说一个我在生产环境吃过亏的细节。第一次搭聚合时我为了好看直接在汇总端写了%{host}分流目录结果上游有一条链路没有打header部分Event写入时路径拼接异常数据散落到了一个奇怪的目录里花了大半天才定位到原因。正确的做法是分两步走先用logger sink或者file_roll这类本地sink跑通整条链路在日志里确认headers里确实有host这个字段确认无误后再切到HDFS Sink并启用%{host}路径。所有上游Agent都要检查拦截器是否配置到位因为只要有一路漏打汇聚端的目录就会出现残缺。我个人在实际操作中的体会是聚合配置本身并不复杂Complicated的不是某个参数而是对全局链路的理解——事务边界在哪里、数据可能在哪一阶段滞留、异常发生时该看哪个计数器。把这几个问题想清楚再回头去看聚合的配置你会发现每个参数都各有其位改起来也胸有成竹。这条链路跑顺之后下一个值得研究的就是负载均衡和故障转移了它们和聚合组合在一起才是一套真正能支撑生产环境的完整采集架构。