一次flume exec source采集日志到kafka因为单条日志数据非常大同步失败的踩坑带来的思考
- 2019 年 11 月 7 日
- 笔记
本次遇到的问题描述，日志采集同步时，当单条日志（日志文件中一行日志）超过2M大小，数据无法采集同步到kafka，分析后，共踩到如下几个坑。
1、flume采集时，通过shell+EXEC（tail -F xxx.log 的方式） source来获取日志时，当单条日志过大超过1M时，source端无法从日志中获取到Event。
2、日志超过1M后，flume的kafka sink 作为生产者发送给日志给kafka失败，kafka无法收到消息。
以下针对踩的这两个坑做分析，flume 我使用的是1.9.0版本。 kafka使用的是2.11-2.0.0版本
问题一、flume采集时，通过shell+EXEC（tail -F xxx.log 的方式） source来获取日志时，当单条日志过大超过1M时，source端无法从日志中获取到Event。flume的配置如下：
...... agent.sources = seqGenSrc ...... # For each one of the sources, the type is defined agent.sources.seqGenSrc.type = exec #agent.sources.seqGenSrc.command = tail -F /opt/logs/test.log|grep businessCollection|awk -F '- {' '{print "{"$2}' agent.sources.seqGenSrc.command = tail -F /opt/logs/test.log|grep businessCollection agent.sources.seqGenSrc.shell = /bin/bash -c agent.sources.seqGenSrc.batchSize = 1 agent.sources.seqGenSrc.batchTimeout = 90000 ......
原因：采用shell+EXEC方式的时候，flume的源码中使用的是如下的方式来获取日志
private Process process = null; //使用这种方式来执行命令。 process = Runtime.getRuntime().exec(commandArgs); //读取日志 reader = new BufferedReader( new InputStreamReader(process.getInputStream(), charset));
在一行日志超过1M后，这个代码就假死了，一直宕住，导致无法获取到数据。
针对这个问题处理方式：
方式一:修改源码的实现方式。（1.9.0的源码 对应的是源码中的flume-ng-core 项目中的org.apache.flume.source.ExecSource.java 这个类）
//process的采用如下方式获和执行命令，就改一行代码。增加.redirectErrorStream(true)后，输入流就都可以获取到，哪怕超过1M process = new ProcessBuilder(commandArgs).redirectErrorStream(true).start();
修改完成后，重新打包编译，然后将生成的jar包替换原来老的jar包。
方式二：放弃EXECSource，使用TAILDIR Source。 使用这个source时，对应的配置如下：
...... agent.sources = seqGenSrc ...... # For each one of the sources, the type is defined agent.sources.seqGenSrc.type = TAILDIR agent.sources.seqGenSrc.positionFile = ./taildir_position.json agent.sources.seqGenSrc.filegroups = seqGenSrc agent.sources.seqGenSrc.filegroups.seqGenSrc = /opt/logs/test.log agent.sources.seqGenSrc.fileHeader = false agent.sources.seqGenSrc.batchSize = 1 ......
建议采用TAILDIR Source 比较好，这个可以对多个日志进行监控和采集，而且日志采集时会记录日志采集位置到positionFile 中，这样日志采集不会重复。EXEC SOURCE在重启采集时数据会重复采集，还需要其他的方式去避免重复采集
问题二、日志超过1M后，flume的kafka sink 作为生产者发送给日志给kafka失败，kafka无法收到消息
原因：kafka 在默认情况下，只能接收1M大小以内的消息，在没有做自定义设置时。所以单条消息大于1M后是无法处理的。
处理方式如下：
1）、修改kafka 服务端server.properties文件，做如下设置(修改大小限制)
# The send buffer (SO_SNDBUF) used by the socket server socket.send.buffer.bytes=502400 # The receive buffer (SO_RCVBUF) used by the socket server socket.receive.buffer.bytes=502400 # The maximum size of a request that the socket server will accept (protection against OOM) socket.request.max.bytes=104857600 message.max.bytes=5242880 replica.fetch.max.bytes=6291456
2）、修改producer.properties，做如下设置(修改大小限制)
# the maximum size of a request in bytes max.request.size= 9242880
3）、java代码中在初始化kafka 生产者时，也需要指定max.request.size= 9242880
Properties properties = new Properties(); ... properties.put("max.request.size", 5242880); ... KafkaProducer<Object,Object> kafkaProducer = new KafkaProducer<Object,Object>(properties);
4)、消费者在消费kafka数据时，也需要注意设置消费消息的大小限制
Properties properties = new Properties(); ... properties.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 6291456); ... KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
对于flume不了的同学，可以看flume 1.9中文版用户指南：https://www.h3399.cn/201906/700076.html
