flume监控文件写入kafka

flume监控日志写到 Kafka，我为什么不用 log4j2 直连

日志采集有一个常见的争议：应用到底该直接把日志打到 Kafka，还是先写本地文件再让 Flume 采？

我之前也纠结过这个问题。后来线上出过一次事故——Kafka 集群抖动，log4j2 的 Kafka appender 直接把应用线程堵死了，服务全挂。从那以后我就认了：应用只管写文件，采集的事交给 Flume。

这篇我就讲一下 Flume 监控文件写到 Kafka 的配置方法，以及几个我用下来觉得关键的参数。

为什么中间要加一层 Flume？

直接让应用通过 log4j2 写 Kafka，确实省事，配置几行就搞定。但带来的问题是：

• Kafka 出问题会影响应用。Broker 不可用、网络抖动、topic 被删，应用的生产者线程会阻塞，业务接口跟着超时。
• 应用要关心 Kafka 的配置。bootstrap.servers、序列化方式、acks 这些参数都塞到应用的配置文件里，跟业务配置混在一起。
• 没法做数据清洗。log4j2 的 appender 只负责发，你想在写入前做格式转换、过滤敏感信息，基本没法搞。

加了 Flume 这一层之后：

• 应用只写本地文件，write 完就结束，Kafka 死活跟它没关系
• Flume 负责采、负责发、负责重试，采集挂了也不影响业务
• 想清洗数据？加个拦截器就行，随时改随时生效

实战配置：从文件监控到 Kafka 写入

先用一个最简单的方案把流程跑通：Exec Source 监控一个文件，Memory Channel 暂存，Kafka Sink 发到 Kafka。

# 1. 定义组件名称（Agent、Source、Channel、Sink）  

#事件源名称
agent.sources = execSource
#通道名称
agent.channels = memoryChannel
#接收器名称
agent.sinks = kafkaSink

# 2. 配置 Source：监控本地文件（以 Exec Source 为例）  
# For each one of the sources, the type is defined
agent.sources.execSource.type = exec
# 执行 tail -F 命令监控日志文件（实时追踪新增内容）  
agent.sources.execSource.command = tail -F /var/log/app/app.log  
# 命令执行失败后自动重启（确保高可用）  
agent.sources.execSource.restart = true  
# 重启间隔（毫秒）
agent.sources.execSource.restartThrottle = 3000  

# 3. 配置 Channel：内存通道暂存数据  
# The channel can be defined as follows.
# 事件源的通道，绑定通道
agent.channels.memoryChannel.type = memory
# 最大缓存事件数（根据内存调整）
agent.channels.memoryChannel.capacity = 1000  
# 每次事务处理的最大事件数  
agent.channels.memoryChannel.transactionCapacity = 100  


# 4. 配置 Sink：写入 Kafka 主题  
# Each sink's type must be defined
# kafka接收器配置
agent.sinks.kafkaSink.type = org.apache.flume.sink.kafka.KafkaSink
# Kafka 集群地址（多个 broker 用逗号分隔）  
agent.sinks.kafkaSink.kafka.bootstrap.servers = localhost:9092
# 目标 Kafka 主题（需提前创建）  
agent.sinks.kafkaSink.kafka.topic = flume-kafka
# 消息序列化方式（字符串序列化）  
agent.sinks.kafkaSink.kafka.serializer.class = kafka.serializer.StringEncoder
# 生产者确认机制：1 表示至少一个副本写入成功（平衡可靠性和性能）  
agent.sinks.kafkaSink.kafka.producer.acks = 1
# 批量写入大小（字节）：积累到 16KB 再发送，减少网络请求  
agent.sinks.kafkaSink.kafka.producer.batch.size = 16384  
# 批量发送延迟（毫秒）：若 500ms 内未达 batch.size，也触发发送  
agent.sinks.kafkaSink.kafka.producer.linger.ms = 500 

# 5. 绑定组件关系（核心！连接 Source → Channel → Sink）  
#Specify the channel the sink should use
# Source 输出到 Channel
agent.sinks.kafkaSink.channel = memoryChannel

# Sink 从 Channel 读取数据 
agent.sinks.kafkaSink.channel = memoryChannel   

关键参数解析

配置文件中以下参数直接影响可靠性和性能，需重点关注：

组件	参数	作用与建议值
Source	restart = true	命令失败后自动重启，确保监控不中断
Channel	capacity = 10000	Memory Channel 的 capacity 设太小，高峰流量直接撑爆；设太大，占内存。我一般先估算每秒多少条日志，乘以 3-5 秒的缓冲量，比如每秒 200 条，capacity 设 1000 就够了。
Sink	kafka.producer.acks = 1	可靠性配置：0（最快）、1（平衡）、-1（最可靠）
Sink	batch.size + linger.ms	batch.size：攒到多少字节再发，默认 16KB linger.ms：攒不够 batch.size 的话，等多久强制发，默认 0 表示立刻发 这两个是”二选一触发”的关系。我一般把 batch.size 调大到 64KB，linger.ms设 500ms，吞吐能上来不少，延迟也就多个几百毫秒，日志采集完全能接受。

启动 Flume Agent命令

执行以下命令启动 Flume，开始监控文件并写入 Kafka：

flume-ng agent \  
  -c /usr/local/Cellar/flume/1.9.0_1/libexec/conf \  # Flume 配置目录（含 flume-env.sh）  
  -f conf/flume-file-to-kafka.conf \  # 自定义配置文件路径  
  --name agent \  # Agent 名称（需与配置文件中一致）  
  -Dflume.root.logger=INFO,console  # 可选：控制台输出日志，便于调试  

验证数据写入 Kafka

通过 Kafka 命令行工具验证数据是否成功写入：

方法 1：消费 Kafka 主题

# 启动 Kafka 消费者，监听 flume-kafka 主题  
kafka-console-consumer.sh \  
  --bootstrap-server localhost:9092 \  
  --topic flume-kafka \  
  --from-beginning  # 从头消费所有数据  

若配置正确，消费者会实时输出日志文件中的新增内容。

方法 2：查看 Kafka 日志文件

Kafka 消息物理存储在日志文件中，可通过以下命令查看：

# 查看主题分区日志（需替换实际日志路径）  
kafka-run-class kafka.tools.DumpLogSegments \  
  --files /usr/local/var/lib/kafka-logs/flume-kafka-0/00000000000000000000.log \  
  --print-data-log  

输出中 payload 字段即为 Flume 写入的日志内容，例如：

payload: "2024-07-22 10:00:00 [INFO] User login success: user_id=123"  

生产环境别用这套配置，得换几个组件

1. 替换 Source 为 Taildir Source

Flume 挂了再起来，tail -F 会从文件开头重新读，数据重复，生产环境建议使用 Taildir Source 监控文件，支持断点续传：

# 替换 Source 配置为 Taildir Source  
agent.sources.execSource.type = TAILDIR  
# 监控的文件路径（支持通配符）  
agent.sources.execSource.filegroups = log1  
agent.sources.execSource.filegroups.log1 = /var/log/app/*.log  
# 偏移量记录文件（重启后从断点继续）  
agent.sources.execSource.positionFile = /var/flume/taildir_position.json  

这个 Source 会把每个文件的读取进度记到 positionFile 里，重启之后接着读，不重不漏。

2. 使用 File Channel 增强可靠性

Memory Channel 在 Flume 崩溃时会丢失数据，对可靠性要求高的场景建议使用 File Channel：

# 替换 Channel 配置为 File Channel  
agent.channels.memoryChannel.type = file  
agent.channels.memoryChannel.checkpointDir = /var/flume/checkpoint  # 元数据目录  
agent.channels.memoryChannel.dataDirs = /var/flume/data  # 数据存储目录（多路径用逗号分隔）  
agent.channels.memoryChannel.capacity = 100000  # 最大事件数  

3. Kafka 生产者参数调优

根据业务需求调整 Kafka 生产者参数，平衡性能与可靠性：

# 提高吞吐量：增大批量发送大小和缓冲区  
agent.sinks.kafkaSink.kafka.producer.batch.size = 65536  # 64KB  
agent.sinks.kafkaSink.kafka.producer.buffer.memory = 67108864  # 64MB  

# 网络优化：设置超时时间  
agent.sinks.kafkaSink.kafka.producer.retries = 3  # 重试次数  
agent.sinks.kafkaSink.kafka.producer.request.timeout.ms = 30000  # 请求超时  

另外：Flume 启动的时候不会帮你创建 topic（会报错 Topic flume-kafka does not exist），得提前建好

kafka-topics.sh --bootstrap-server localhost:9092 --create --topic flume-kafka --partitions 3 --replication-factor 1  

4. 日志清洗与转换

Flume 的拦截器可以在 Source 往 Channel 发之前，对 Event 做加工。加时间戳是最常见的用法：

# 配置拦截器：添加时间戳头信息  
agent.sources.execSource.interceptors = timestampInterceptor  
agent.sources.execSource.interceptors.timestampInterceptor.type = org.apache.flume.interceptor.TimestampInterceptor$Builder  

这样每个 Event 的 Header 里会多一个 timestamp 字段，下游消费的时候可以用来做时间窗口聚合。

如果你想过滤掉 DEBUG 级别的日志，或者脱敏手机号，也可以写自定义拦截器，实现 Interceptor 接口就行。

5. 保证日志文件权限

错误提示：Permission denied: /var/log/app/app.log
解决：确保 Flume 进程对监控文件有读权限，或修改文件权限：

chmod 644 /var/log/app/app.log