flume读取文件延迟调优

问题：

需要将数据库A的数据同步给数据库B。通过采集A库的sql操作日志，在B库中执行。采集A库时的flume读取日志比日志生成时间延迟，且延迟时间递增。

解决：

i3使用自定义正则表达式过滤器，进行数据过滤。

自定义正则表达式过滤器：CustomRegexFilteringInterceptor

使用matches()方法匹配：

Pattern pattern = Pattern.compile(regrex);
if(pattern.matcher(content).matches()){
    return event;
}

将flume-kafka-conf.properties中i3的设置改为：

agentkafka.sources.sr1.interceptors.i3.type = com.ai.csc.boss.flume.interceptor.CustomRegexFilteringInterceptor$Builder

原有：flume-kafka-conf.properties配置如下：

agentkafka.sources = sr1
agentkafka.channels = c1
agentkafka.sinks = sk1

# For each one of the sources, the type is defined
agentkafka.sources.sr1.type = TAILDIR
agentkafka.sources.sr1.channels = c1
agentkafka.sources.sr1.positionFile = ../bin/agentkafka_taildir_position.json
agentkafka.sources.sr1.filegroups = f1
# Location of postgresql log files
agentkafka.sources.sr1.filegroups.f1 = /app/flume/postgresql-20.*csv
agentkafka.sources.sr1.fileHeader = true
agentkafka.sources.sr1.inputCharset = utf-8

#拦截器定义，只保留insert、update、delete类型的日志
agentkafka.sources.sr1.interceptors = i1 i2 i3
agentkafka.sources.sr1.interceptors.i1.type = regex_filter
# Database name of source DB
agentkafka.sources.sr1.interceptors.i1.regex = .*?,\"abcd\",.*
agentkafka.sources.sr1.interceptors.i1.excludeEvents = false
agentkafka.sources.sr1.interceptors.i2.type = regex_filter
agentkafka.sources.sr1.interceptors.i2.regex = .*?(execute.*: |statement: )(insert|update|delete|INSERT|UPDATE|DELETE).*
agentkafka.sources.sr1.interceptors.i2.excludeEvents = false
agentkafka.sources.sr1.interceptors.i3.type = regex_filter
# schema_name and table name which should be ignored
agentkafka.sources.sr1.interceptors.i3.regex = 大堆需要排除的表（大小写）的正则表达式
agentkafka.sources.sr1.interceptors.i3.excludeEvents = true

agentkafka.channels.c1.type = memory
agentkafka.channels.c1.keep-alive = 10
agentkafka.channels.c1.capacity = 100000
agentkafka.channels.c1.transactionCapacity =10000

agentkafka.sinks.sk1.channel = c1
agentkafka.sinks.sk1.type = com.flume.sink.kafka.KafkaSink
# target topic in kafka cluster
agentkafka.sinks.sk1.kafka.topic = TOPIC_ABC
# target ip:port list of kafka cluster
agentkafka.sinks.sk1.kafka.bootstrap.servers = ip:port
agentkafka.sinks.sk1.kafka.producer.key.serializer = org.apache.kafka.common.serialization.LongSerializer
agentkafka.sinks.sk1.kafka.flumeBatchSize = 2000
agentkafka.sinks.sk1.kafka.producer.acks = 1
agentkafka.sinks.sk1.kafka.producer.linger.ms = 1
agentkafka.sinks.sk1.kafka.producer.compression.type = snappy

原因：

正则匹配方法中，find()方法是部分匹配，是查找输入串中与模式匹配的子串；matches()方法是全部匹配，是将整个输入串与模式匹配。过滤大量数据时，matches()比find()方法更快。

flume的RegexFilteringInterceptor使用find()方法匹配字符串。

以下文字可以忽略

好吧，我承认这是一次漫长而且失败的问题查找经历，尽管问题最终得到解决。

为了查明原因，首先开启了flume的http监控。启动命令增加：-Dflume.monitoring.type=http -Dflume.monitoring.port=1234

flume的监控可参考https://www.jianshu.com/p/09493efe0fb8或者自行百度，顺便自行搜索“http监控性能指标”

然后，安装grafana + influxdb + telegraf，进行数据收集展示

参考链接中telegraf的telegraf.conf中使用的是 [[inputs.httpjson]]，但是根据官网，1.6及以上已经不用了，具体请查官网。所以该配置中使用[[inputs.http]]

 [[inputs.http]]
	urls = [
		"http://ip:port/metrics"
	]
	method = "GET"
	timeout = "1s"
	json_name_key = "SOURCE.sr1_Type"
	json_string_fields = ["SOURCE.sr1_EventReceivedCount","SOURCE.sr1_EventAcceptedCount","SOURCE.sr1_AppendBatchReceivedCount","SOURCE.sr1_AppendReceivedCount"]
	data_format = "json"

参考：https://kiswo.com/article/1023
http://blog.51cto.com/11512826/2056183

接下来才是浪费时间的阶段：用开发库的日志测试，读取很速度，不解为啥现场很慢。获取现场日志，进行测试，速度下降。ok，至少问题复现了。通过top命令查看flume的cpu很高（参考https://blog.csdn.net/linux_ja/article/details/47276771），主要是

org.apache.flume.interceptor.RegexFilteringInterceptor.intercept(RegexFilteringInterceptor.java:104)

当然这个时候，还觉得不可思议，遂将i1-i3进行逐个修改拦截，发现i3注释掉之后，速度陡升。

终于开始查看flume的源码，通过对find()和matches()的方法测试，发现原因所在。

最后的最后，其实，使用flume的过滤器时，i3的正则表达式里，过滤了十几张表的大小写，当缩小的只有4张表的大写时，速度也是有很大的提升的。

注：日志文件50M左右，实际从sink发送的数据10k左右。

And，我想接下来，大概要找我调整cpu了，因为之前差不多是100%。哭，大哭，嚎啕大哭。

在写完以上文字1h内，把i1、i2也改成自定义过滤器之后，觉得可以交差了。

50M的文件，从最初14min读取17M左右，到（缩小i3表达式或者使用自定义过滤器）8min读完，到将i1\i2\i3使用自定义过滤器30s读完。

banyanmars

flume读取文件延迟调优