1. Flume是什么先建立整体数据观做了多年数据平台建设我发现很多团队第一次接触Flume时都习惯把它理解成“一个收集日志的工具”。这个说法没错但容易让人低估它和其他简单采集脚本之间的差距。这篇Flume入门及进阶教程我想按我平时带新人的思路来讲先建立整体直觉再拆组件选型然后直接上可复现的配置最后把线上最容易踩的坑一个个摆出来。这样无论是刚接触采集链路的新人还是已经部署过但经常被故障折腾的运维同学都能找到自己需要的东西。Flume是一个分布式的、可靠的日志聚合系统核心能力是把分散在多台机器上的日志文件、网络数据、消息队列数据按照你定义的规则汇聚起来落到HDFS、Kafka、Hive或者其他存储中。它最典型的定位是“大数据链路的第一公里”也就是数据从业务服务器出发进入数仓或实时链路之前的那一段搬运工作。用物流来类比的话Source是收货口Channel是临时仓库Sink是送货的货车三者协调才能保证货物不断流、不堵死。1.1 日志从机器到数仓Flume到底解决什么问题先说业务场景。假设有10台业务服务器每台都在不断追加写入app.log这些日志需要进到数仓做分析。没用Flume之前不少人会写个定时任务用scp把文件整个拉走或者用rsync同步。这种方式在日志量小的时候也能跑但很快就会遇到几个麻烦文件持续增长每次全量拷贝浪费带宽增量拷贝要自己记录偏移量进程重启就乱文件一旦被rotate很容易漏掉中间新生成的日志传输失败后没有重试和断点续传的概念数据丢失了也不知道。Flume解决的就是这批脏活。它允许你在每台业务机上部署一个AgentAgent用taildir这类Source去监听一个或多个日志文件只读取文件新增的内容并把偏移量记录到本地positionFile。随后这批数据被封装成Event通过Channel缓冲再由Sink写入目标系统。整个过程里Source负责“怎么读”Channel负责“怎么存”Sink负责“怎么写”三者职责分离改动任何一个环节都不需要推翻整套流程。我还常被问到Flume和Filebeat、Logstash有什么区别。这几个工具确实有部分功能重叠但侧重点不同。Flume的优势在于和Hadoop生态结合得非常紧密HDFS Sink可以直接按时间、按分区生成目录支持文件滚动、压缩、批量提交配置得当的话基本不用再写额外的落地脚本。Filebeat轻量但偏向纯日志传输Logstash强在过滤加工但常驻内存开销更大。入门阶段不用纠结谁取代谁如果你明确要把日志“源源不断搬进HDFS或Kafka”Flume是相当顺手的方案。1.2 Agent、Source、Channel、Sink以及一条Event的完整旅程Flume的部署单元叫Agent一个Agent就是一个独立进程。Agent内部包含Source、Channel、Sink三类核心组件配置文件里通过命名来绑定它们之间的关系。 Source负责对接外部数据来源比如taildir读文件、avro接收上游Agent数据、netcat监听端口。Channel是Source和Sink之间的缓冲区目的是解耦读写速度防止Sink一慢就把Source堵死。Sink负责把数据写到目标端比如HDFS、Kafka、Hive。在Flume里一条被传输的数据叫Event结构是“headers body”。body就是实际内容通常是一行日志的字节数组headers是键值对像timestamp、hostname这类元信息都可以塞在里面。headers的价值很大HDFS Sink路径中的日期、主题分区以及多路转发时的路由判断都依赖headers。一条Event的完整旅程是这样的Source读到数据后调用ChannelProcessor把Event放到一个事务里再提交给Channel这叫put事务Sink从Channel里取数据时同样开启一个事务把Event读出来成功写入目标后提交这叫take事务。两者都走“开启事务、写入、提交”的流程所以Flume对外承诺的是At Least Once也就是至少一次语义。这意味着正常情况下数据不丢但极端故障下可能出现重复下游要根据业务做幂等或去重。我第一次听到“两个事务”时觉得有点绕但后来排查丢数据问题时才明白Channel事务才是理解Flume可靠性的钥匙。后面第5章我会深入展开这里先记住一件事Flume不是把数据放进Channel就完了Sink成功提交事务后才算真正完成一次搬运。2. 组件选型拆解Source、Channel、Sink如何搭配才不踩坑2.1 Source选型读文件怎么选别让Exec成为埋坑点Source的类型非常多我把实际工作中最常用的列成一张表方便对比Source类型适用场景关键注意点taildir监控多个追加写入的日志文件支持断点续传推荐但要求文件内容以追加方式写入spooldir监控一个目录目录中出现的新文件被采集文件必须原子地放入目录不能边写边放exec执行shell命令读取输出比如tail -F进程重启后命令重新执行容易丢数据或重复不推荐avro/thrift接收上游Agent或应用SDK发送过来的数据常用于多级Agent串联netcat/http测试或轻量接入适合临时调试不适合生产大流量场景kafka source从Kafka topic消费数据需要维护offset常用于多路汇聚taildir是我在生产环境的主力选择。它用一个JSON格式的positionFile记录正在监控文件的inode和读取偏移量Agent重启后能接着上次的位置继续读。这个能力非常重要因为日志采集进程最怕的就是重启后从头开始读老文件或者完全忘记自己读到哪。我对exec类型的Source保持一定戒心。网上很多老教程会拿exec配合tail -F做演示看起来简单但ExecSource一旦进程重启tail命令会重新执行相当于把文件重新读一遍重复数据就这么产生了。生产中如果监听文件追加内容优先选taildir如果监听的是固定目录中不断产生的新文件选spooldir但放入目录的文件必须命名成不会再被修改的状态。2.2 Channel选型Memory、File、Kafka怎么取舍Channel是传输链路上最容易被忽视、也最影响稳定性的组件。选错Channel类型后面大概率会付出丢数据或性能劣化的代价。我一般按这张表来做初步判断Channel类型性能可靠性适用场景MemoryChannel极高纯内存操作进程崩溃或宕机时数据可能丢失对丢少量数据不敏感追求吞吐FileChannel中等需要写磁盘通过WAL机制持久化可靠性高日志重要不能随意丢失KafkaChannel取决于Kafka集群数据在Kafka中可靠性高需要解耦、已引入Kafka的环境MemoryChannel配置简单速度最快但关键问题是劫机或进程被kill时Channel里还没被Sink取走的数据会全部丢失。很多团队“莫名其妙丢数据”的根源就是从日志看Sink确实在写但Agent一重启就丢掉一大段正是MemoryChannel在作怪。FileChannel把事件写入本地磁盘的WAL文件可靠性高得多代价是吞吐量下降、磁盘IO占用上升。如果数据重要我会直接上FileChannel同时单独给它分配磁盘空间不要和系统盘抢IO。KafkaChannel则是把Kafka当作Channel的存储层数据先落到Kafka再由Sink消费出来写到最终目的地。它适合那些不想在业务机上多维护一套存储、又希望数据有更强持久性的场景。需要明确的是KafkaChannel和Kafka Sink不是一回事后面进阶部分我会专门对比。配置Channel时有两个参数必须理解capacity和transactionCapacity。capacity是Channel最多能缓冲多少EventtransactionCapacity是单个事务最多能处理的Event数量。按照官方要求transactionCapacity必须小于等于capacity否则配置加载时就会报错。实际使用中transactionCapacity还要和Source的batchSize、Sink的batchSize配合否则小事务不断刷Channel反而拖慢吞吐。2.3 Sink选型写HDFS、Kafka还是先落本地文件Sink决定数据最终去向。多数场景下HDFS Sink和Kafka Sink使用频率最高我把它们和一些辅助类型放在一起说明Sink类型目标常见用途HDFS SinkHDFS文件数据落地数仓按时间目录滚动Kafka SinkKafka topic接入实时链路或作为缓冲Hive SinkHive分区表直接写入Hive分区目录File Roll Sink本地文件先落本地磁盘再由其他任务搬运Avro Sink下游Flume Agent多级Agent串联Logger Sink控制台日志调试配置时非常好用HDFS Sink的配置项看起来很多但核心就三件事文件写到哪个目录、文件多久滚动一次、一批写多少个Event。路径里可以用%Y%m%d、%H这类时间占位符也可以用%{headerName}引用Event的headers这个能力在做小时级分区或按业务主题分区时很实用。滚动策略由rollInterval、rollSize、rollCount三个参数控制分别是“按时间滚动”“按文件大小滚动”“按Event数量滚动”。默认情况是满足任意一个条件就关闭当前文件新建下一个文件。生产中容易出问题的地方是很多人只调大了rollSize却忘了rollInterval默认是30秒结果小流量场景下每几十秒就生成一个空文件HDFS上冒出一堆小文件。后面我会把这套参数的调法单独拿出来讲。Kafka Sink则简单很多通常只需要指定bootstrapServers和topic。它适合把数据送到实时计算平台但如果你已经选了KafkaChannel就不需要再配Kafka Sink了。2.4 组合原则先跑通最小闭环再谈优化和扩展我见过不少新手同学第一次配置就想把Source、Selector、Interceptor、多路Channel、负载均衡Sink一次性配好。结果配置文件写得非常长启动后一个问题叠一个问题最后连数据到底有没有被Source读到都说不清楚。更好的做法是永远先搭一个最小闭环一个Source、一个Channel、一个Sink先让数据完整地从A到B再逐个叠加功能。选型上我建议默认组合是taildir Source FileChannel HDFS Sink。如果是实时链路则换成taildir Source MemoryChannel或KafkaChannel Kafka Sink。生产环境里我会尽量避免把MemoryChannel的capacity设得过大因为它只是缓冲不是持久化仓库设得再大也只是把丢失窗口变大。更合适的思路是给足能力让Sink及时消费让Channel保持一个健康水位。还有一个容易被忽略的原则Agent进程不要承担太多职责。很多人喜欢在一个Agent里塞多个Source和多个Sink看着省事但一旦其中一个Sink故障整个Agent的Channel可能被拖慢波及所有数据流。我的建议是同一类用途的数据流用一个独立的Agent实例来跑配置清晰故障影响面也小。3. 动手实操从零配置一套日志采集到HDFS的完整链路3.1 环境准备与基础验证Flume依赖Java运行环境我在实际部署中用的是JDK 1.8系列的版本兼容性比较稳。拿到Flume安装包后直接解压到统一目录比如/opt/apache-flume然后把bin目录加进PATH。环境准备好后先验证一下安装是否正常执行命令bin/flume-ng version如果能正常输出版本信息说明JDK和FLUME_HOME基本没问题。启动Agent的命令则是这样bin/flume-ng agent --name agent1 --conf conf -f conf/flume-taildir-hdfs.conf -Dflume.root.loggerINFO,console这里有个容易踩的坑修改配置文件后Agent并不会自动加载新配置必须重启进程才会生效。很多同学改了文件半天看不到效果其实只是Agent还在用旧的配置运行。调试阶段建议用-Dflume.root.loggerINFO,console把日志打到控制台方便实时观察生产环境则建议打成文件日志避免前台启动方式把进程挂死在终端上。3.2 最常用的taildir到HDFS完整配置下面这份配置是我在模拟项目X中反复用过的模板结构足够简单但覆盖了生产常用的要点agent1.sources s1 agent1.channels c1 agent1.sinks k1 agent1.sources.s1.type taildir agent1.sources.s1.positionFile /opt/flume/pos/taildir_position.json agent1.sources.s1.filegroups f1 agent1.sources.s1.filegroups.f1 /data/logs/app/.*\\.log$ agent1.sources.s1.headers.f1.topic app_log agent1.channels.c1.type file agent1.channels.c1.checkpointDir /opt/flume/checkpoint agent1.channels.c1.dataDirs /opt/flume/data agent1.channels.c1.capacity 100000 agent1.channels.c1.transactionCapacity 5000 agent1.sinks.k1.type hdfs agent1.sinks.k1.hdfs.path /user/flume/events/%{topic}/%Y%m%d/%H agent1.sinks.k1.hdfs.filePrefix app_log_ agent1.sinks.k1.hdfs.fileType DataStream agent1.sinks.k1.hdfs.writeFormat Text agent1.sinks.k1.hdfs.rollInterval 3600 agent1.sinks.k1.hdfs.rollSize 268435456 agent1.sinks.k1.hdfs.rollCount 0 agent1.sinks.k1.hdfs.batchSize 1000 agent1.sources.s1.channels c1 agent1.sinks.k1.channel c1配置里值得注意的有这么几处。positionFile的目录必须先创建否则首次启动可能因为无法创建文件而报错。filegroups可以配置多组正则每组还可以通过headers.xxx设定一个默认header这里我把topic设为app_log这样Sink的路径参数%{topic}就能被替换。FileChannel的checkpointDir和dataDirs是独立目录启动前同样要确保存在。HDFS Sink的path中使用%Y%m%d/%H意味着数据会按天、按小时自动分目录这是数仓分区最常用的格式。rollInterval设置成3600秒rollSize设置成256MBrollCount设置为0表示不按Event数量滚动。实际中这些参数要根据数据量来调整后面故障排查那一章我会展开。配置写好后往日志文件里追加几行数据再用Logger观察Agent日志或者直接去HDFS目录看生成的文件。第一次看到HDFS上出现带.tmp临时后缀的文件时不用慌那是Flume正在写入的文件等Sink完成一批写入并关闭后临时文件会重命名成正式文件。这个机制保证下游看不到写了一半的文件。3.3 多路复用与扇出一份数据同时进两套系统有时候一份日志既要进HDFS做离线分析又要进Kafka做实时计算。最容易想到的方案是部署两个Agent分别采集同一份文件但这样会重复读取日志浪费IO还可能出现两个Agent读同一文件的偏移量冲突。更规范的方案是在一个Agent里用一个Source把Event复制到多个Channel再由不同Sink消费这就是Flume的扇出。默认情况下一个Source可以配置多个ChannelEvent会被复制到所有Channel这叫replicating selector。如果你想根据Event内容决定发到哪个Channel比如只把错误日志发给在线告警链路其他日志正常进数仓那就要用multiplexing selector。配置大致像下面这样agent1.sources.s1.selector.type multiplexing agent1.sources.s1.selector.header log_level agent1.sources.s1.selector.mapping.ERROR c_error agent1.sources.s1.selector.default c_normal agent1.sources.s1.channels c_error c_normalmultiplexing的判断依据是Event headers里的某个字段。比如用regex_extractor拦截器从日志中提取log_level塞进headersSelector就能根据这个值把Event路由到不同Channel。这个能力在复杂链路里非常实用可以帮我们省下一套额外的日志解析服务。扇出模式的一个性能提示是数据被复制意味着写入量翻倍下游Sink的消费能力要跟上否则Channel会积压。我曾经因为配置了一个Source复制到三个Channel结果其中两个HDFS Sink都写得很慢直接把第三个Kafka链路也拖住了。后来我严格控制扇出路数或者改用KafkaChannel做中转问题才缓解。3.4 用Interceptor完成简单的数据加工Flume的拦截器Interceptor能在Event进入Channel之前做轻量处理常用在内置header、过滤无用日志、提取关键字段这几个场景。比如只采集包含ERROR的日志行可以使用regex_filter拦截器agent1.sources.s1.interceptors i1 agent1.sources.s1.interceptors.i1.type regex_filter agent1.sources.s1.interceptors.i1.regex ERROR agent1.sources.s1.interceptors.i1.excludeEvents false上面的配置会保留能匹配到ERROR的Event。如果设置excludeEvents为true作用正好反过来变成丢弃匹配的Event。还有regex_extractor拦截器它可以从日志正文中用正则捕获某一部分内容写入Event的headers比如提取用户ID、日志级别、接口名。后面HDFS路径里用%{user_id}做分区时这种提取方式非常方便。拦截器的处理逻辑每一条Event都会走一遍所以正则表达式要尽量精简不要写那种把所有内容都捕获一遍的巨型正则。拦截器数量也不要贪多我曾经见过一个Agent里串了五个拦截器做各种解析每个拦截器都消耗CPU直接把整个采集链路的吞吐拉低了一半。能不做加工就别做Flume的核心职责是把数据稳定搬走复杂清洗应该在数仓或流计算阶段去做。4. 线上常见问题与排错手段实录4.1 Source明明有数据但Sink一直不落盘这是我最常被问到的问题。日志文件在增长控制台也没看到异常但HDFS或Kafka里就是没有新数据。遇到这种情况我第一反应不是去看Source而是看Sink。Sink一旦写失败会反复重试并把Channel里的数据一直押着不提交表现上就像“Source卡住了”。正确排查顺序是先看Agent日志里有没有Sink报错再看Channel是否被占满。实际操作中日志里最常出现的HDFS异常有两类。一类是权限问题报类似Permission denied的错一般是Agent启动用户的权限不够或者Kerberos票据没有正确初始化另一类是FileSystem连接问题比如默认Hadoop配置没配好Flume根本不知道hdfs://地址怎么解析。把export HADOOP_HOME和ClassPath配好很多HDFS连接问题会直接消失。还有一个隐蔽原因是Channel的事务容量设置不正确。比如capacity设得很大但transactionCapacity比Source单次提交的batch小得多Source会频繁提交失败。出现过这种情况后我养成了一个习惯启动Agent后先看几秒日志确认Source每秒上报的Event数、Sink每批写的Event数再离开终端。刚启动的头几十秒最能暴露配置不对的问题。4.2 Channel背压和吞吐瓶颈Flume的Channel本质是一个有界缓冲区上游Source生产速度快下游Sink消费速度慢Channel就会被填满。填满后Source会阻塞提交数据看似“堆积在源头”其实是Channel在下游施加背压。这个问题排查起来不难通过JMX或监控指标看ChannelFillPercentage就能定位但我更想说的是它背后的容量规划思路。通用做法是先把batchSize和transactionCapacity调大比如把Sink的batchSize从200调到1000让单次写HDFS的吞吐更高。如果还顶不住就增加Sink并行度也就是给同一个Agent搭配多个Sink每个Sink对应独立的Channel分摊写入压力。需要注意的是HDFS Sink的并行写入会让文件数量变多小文件问题会更明显这是一对需要平衡的矛盾。我自己比较推荐的做法是对单Agent吞吐做压测。比如用一个能生成100万行日志的程序把日志写到文件里观察Agent从Source读出来到Sink完成写入的耗时。如果每秒吞吐离预期差太大用jcmd或者监控JSON看看CPU和内存分配情况基本能定位到瓶颈在哪个组件。总的说来Channel不是越大越好它只是一个平滑缓冲区合理的容量应该是“下游故障几分钟内不至于立即丢数据”的缓冲量。4.3 HDFS小文件是Flume采集老生常谈的坑小文件的危害不用多说NameNode内存被大量占用下游Spark或Hive扫描时任务数爆炸。Flume写入HDFS如果配置不当很容易制造出一批小文件。原因通常有三个文件滚动太快、目录分得太碎、单批写入量太小。先说滚动太快的场景。有些默认配置里rollInterval只有30秒如果日志量很小每隔30秒就会生成一个空物理文件。假设你按天数据量只有几百MB日积月累下来HDFS上会堆出几千个文件。解决办法是把rollInterval调大或者直接设为0来关闭时间滚动让文件主要按大小滚动。比如rollInterval0rollSize134217728rollCount0这样只有文件达到128MB才会滚动。如果业务数据量确实很小也可以保留一个较长的滚动时间比如3600秒保证文件不至于一直不关闭。再说目录分太碎的问题。我见过有配置把小时、甚至分钟都写进path像/user/flume/events/%Y%m%d/%H%M结果每个分钟目录里只有一个很小的文件。分区粒度应该以最终消费方的查询粒度为准普遍按天或小时就够了按分钟分区很少能带来收益反而制造一堆小目录。最后还要注意Sink并行度。多个HDFS Sink同时写同一个目录每个Sink都会独立滚动文件文件数量成倍增加。如果为了吞吐加了多个Sink要记得在路径中给每个Sink不同的标识比如filePrefix里带上sink编号否则下游处理时容易乱。4.4 数据丢失与重复怎么缩小影响面Flume对外语义是至少一次也就是正常情况下数据不会丢但可能在进程重启、网络抖动、Sink重试等场景下产生重复。认识到这一点非常重要它决定了我们不该在下游采用简单计数方式来判断数据完整性而要考虑全链路是否具备去重或幂等能力。数据丢失的真正常见原因是用了MemoryChannel。我见过某个线上项目Agent部署在物理机某次机房维护导致机器重启MemoryChannel里积压的几十万条数据全部消失。后来把Channel换成FileChannel同样场景下重启后能继续从本地WAL恢复数据才真正保住。如果公司已有Kafka或者Agent本身就在采集Kafka相关的数据KafkaChannel也能起到类似的持久化效果还能顺带让多个下游独立消费。重复场景常见于Sink写数据成功但提交Channel事务时网络断掉导致同一批数据被Sink重试再次写入。针对HDFS Sink文件关闭前的.tmp机制和批处理机制会降低跨批次重复的概率但并不能做到完全精确一次。所以我在设计方案时会要求下游对每一条数据都具备幂等写入能力比如用日志业务主键做去重表或者接受极少量重复数据由分析侧自行过滤。4.5 常见异常速查与处理建议现象常见原因处理建议启动报找不到Hadoop配置HADOOP_HOME或ClassPath没配置检查环境变量确认Hadoop相关依赖能被Agent加载hdfs://路径无法解析core-site.xml/hdfs-site.xml未加载将Hadoop配置文件放入Agent的conf目录或显式指定ClassPathHDFS写入Permission deniedAgent运行用户权限不足给Agent用户授权对应目录或调整HDFS访问策略positionFile加载失败对应目录不存在或权限不对启动前mkdir确认Agent用户有写权限MemoryChannel容量忽高忽低transactionCapacity设置不合理检查batchSize、transactionCapacity与capacity的关系调小单事务大小Sink写入长时间不提交目标端连接失败Sink在重试查看Sink日志优先解决目标端连接或权限问题文件一直带.tmp后缀Sink还在滚动周期内或写入失败等待滚动条件若长期不关闭则查Sink报错Kafka Sink报topic找不到Kafka版本或topic配置不一致检查kafka.bootstrapServers和kafka.topic配置确认集群可用这张表不是穷尽的但它覆盖了我在实际运维中遇到过的绝大多数低水平事故。我的经验是Flume配置出错百分之六十都发生在最基础的路径、权限、环境变量上而不是所谓的深奥技术问题。排查时先看日志再看监控最后才怀疑源码层面的Bug。5. 往进阶走事务语义、Kafka通道和高可用拓扑5.1 两层事务到底在保护什么前面说过Flume的一条Event要经历Source到Channel的put事务以及Sink从Channel取数据的take事务。理解这两个事务才能真正看懂为什么Flume是“可靠传输”而不是“绝对不丢”。 put事务包含一个批次写入和一次提交就是说Source得把这个批次完整放进去并确认成功才算完成。take事务类似Sink要从Channel里读出一个批次写入目标端成功后再提交Channel才会真正删除这些Event。问题在于Sink和Channel之间的“写入成功”没法做到原子统一。比如HDFS Sink已经写入了文件但本地提交Channel事务时Agent突然崩溃重启后Sink发现原批次还在Channel里于是再次写入HDFS这就出现了重复。反过来如果Source已经提交了put事务但Agent随后崩溃重启后positionFile已经记到了新位置之前进入Channel的数据会继续被Sink消费不会丢。所以至少一次语义是这套事务设计换来的结果。在设计UDP、TCP这类传输模型时我们对“可靠”的要求通常是“不丢不重”但在分布式日志采集里不丢不重实在太贵了。实际业务日志分析场景绝大多数都能容忍少量重复却不能容忍大段丢失。这也是为什么我建议生产链路尽量使用FileChannel或KafkaChannel做一个缓冲层把“待确认数据”先持久化下来给Sink重试争取空间。5.2 Kafka Channel还是Kafka Sink别把两件事搞混Kafka相关组件有两种用法很多人一开始会混淆。第一种Kafka Sink是把Flume中已存在的Event最终写到Kafka topic第二种Kafka Channel是把Kafka当作Channel本身来使用Source把Event写入Kafka topicSink从该topic读取并落到最终目标。两者数据流完全不一样。Kafka Channel的典型拓扑是taildir Source写入KafkaChannel的topic然后配一个HDFS Sink消费这个topic并落HDFS。此时Kafka对于Flume来说既承担了缓冲职责也承担了传输职责。这么做的好处是数据已经持久化在Kafka里即使Sink故障较长时间Kafka也能保留数据同时如果有其他团队希望通过Kafka读取同一份日志他们能直接消费这个topic而不必额外在Flume里再复制一份数据流。代价自然是多了一个Kafka依赖Agent和Kafka集群之间的网络、topic的副本策略都要规划好。Kafka Sink更像一条“数据出口”适合前面用MemoryChannel或FileChannel缓冲、最后写给Kafka的简单场景。两者的选择逻辑本质上是在问你到底需要Kafka帮你解决持久化问题还是只需要它作为下游接收端只要把这个问题想清楚配置就不会乱。5.3 高可用拓扑多级Agent与Sink Group单机部署的Agent总会有单点风险。业务机坏了日志采集自然中断这是没办法的但如果汇聚层的Agent挂了会影响所有下游链路。所以生产环境中我通常会在业务机上部署第一级Agent负责读取本机日志然后通过avro Sink发送到汇聚机上的第二级Agent再由第二级Agent统一写入HDFS或Kafka。这个拓扑叫多层Agent串联。第一级Agent的Source用taildirSink用avro第二级Agent的Source用avroSink用HDFS或Kafka。avro协议是Flume原生的二进制协议效率高也支持事务。配置上非常简单第一级Sink设置hostname和port第二级Source绑定同一个端口监听就行。为了高可用Sink侧还可以配置Sink Group。默认情况下一个Sink绑一个Channel但如果用Sink Group多个Sink可以共同消费一个或多个Channel再由处理器决定怎么分配。Failover模式指定各组Sink的优先级优先Sink挂了自动切换到备用SinkLoadBalance模式则轮流或按权重分发。一个常用的Failover配置如下agent1.sinkgroups sg1 agent1.sinkgroups.sg1.sinks k1 k2 agent1.sinkgroups.sg1.processor.type failover agent1.sinkgroups.sg1.processor.priority.k1 10 agent1.sinkgroups.sg1.processor.priority.k2 5 agent1.sinkgroups.sg1.processor.maxpenalty 30000 agent1.sinks.k1.type hdfs agent1.sinks.k1.channel c1 agent1.sinks.k2.type hdfs agent1.sinks.k2.channel c2Failover模式下每个Sink必须有自己的Channel。比如k1消费c1k2消费c2当k1失败时k2接管k1的Channel要区分“单个Sink Group管理同一Channel”还是“不同Channel”。实际Lang文档说明Failover模式接收“可以与同一个Channel绑定”的多个Sink让我核对Sink Group将多个sink作为单个实体如果Sink Group中的所有sink有一个相同的channel那么Event会被交给其中一个sink。实际上常见配置中一个Sink Group配多个sink都绑定同一个channel。如果使用一个Sink Group配两个sink他们的channel可以都是c1 是的常见教程配置为所有sinkagent1.sinks.k1.channel c1agent1.sinks.k2.channel c1在failover中选择一个sink消费同一个channel。而我上面配置了不同channel。为了准确我应调整sink group中多个sink通常绑定同一个channel。在failover中k1和k2都指向同一个channel优先级决定先让k1消费。如果k1失败k2接管同一个channel。我修正示例agent1.sinkgroups sg1 agent1.sinkgroups.sg1.sinks k1 k2 agent1.sinkgroups.sg1.processor.type failover agent1.sinkgroups.sg1.processor.priority.k1 10 agent1.sinkgroups.sg1.processor.priority.k2 5 agent1.sinks.k1.type avro agent1.sinks.k1.hostname collector1 agent1.sinks.k1.port 4141 agent1.sinks.k1.channel c1 agent1.sinks.k2.type avro agent1.sinks.k2.hostname collector2 agent1.sinks.k2.port 4141 agent1.sinks.k2.channel c1这是一个应用在first agent的向两个collector分发的Failover示例。有效。这样做的好处是业务机上的Agent把日志发给汇聚AA出问题后自动切换到汇聚B下游数据链路不中断。如果你追求更细粒度的负载均衡就把type改成loadbalanceSink之间按轮询或随机策略分数据。不过我要提醒一句LoadBalance只解决并发分发不解决下游重复消费问题得到的数据分散到多个汇聚点后要用时间戳或分区策略保证整体有序性。5.4 监控Agent状态以及后续可以自己动手试的方向部署稳定后不能不看监控。Flume本身提供JMX指标启动Agent时添加如下参数就能通过HTTP拉取监控数据-Dflume.monitoring.typehttp -Dflume.monitoring.port34545启动后访问这个端口能看到ChannelSize、ChannelFillPercentage、Sink SuccessCount、EventDroppedCount等指标。我一般重点关注三个指标ChannelFillPercentage接近100意味着下游跟不上EventDroppedCount非0意味着有数据被主动丢弃Sink SuccessCount长时间不增长意味着Sink卡住或目标端故障。把这些指标接进内部监控大盘后基本可以做到“数据链路挤掉之前先报警”。进阶方向也很多。比如可以试着写自定义Source或Sink接入公司内部的消息中间件可以把拦截器换成自己的子类做更复杂的日志脱敏可以在HDFS Sink中开启压缩编码减少存储成本也可以研究怎么把Channel的磁盘目录和数据目录分散到不同磁盘提高FileChannel的写入性能。每一条都值得用一个小实验来验证而不是直接照抄网上的配置。我个人在实际操作中的一个体会是Flume的难点从来不在于“怎么启动一个Agent”而在于当你面对一个已经跑了半年、不断有小毛病的采集系统时能不能快速判断出问题出在Source、Channel还是Sink。把本章的事务语义和背压机制真正吃透线上排障会顺手很多。最后再分享一个小技巧任何一次配置变更都先在测试目录里用一个Logger Sink跑通数据再切到真实的HDFS或Kafka Sink这能帮你把配置错误和下游故障区分开省下大量复盘时间。