TOP

flume读取文件延迟调优
2019-01-21 02:13:07 】 浏览:293
Tags:flume 读取 文件 延迟

版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/banyanmars/article/details/85157782

问题:

需要将数据库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

吐槽:json格式真是个大坑,大概是英语不好+配置不对吧,反正没理解为啥使用tag_keys配置,始终获取不到数据。

接下来才是浪费时间的阶段:用开发库的日志测试,读取很速度,不解为啥现场很慢。获取现场日志,进行测试,速度下降。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读完。


flume读取文件延迟调优 https://www.cppentry.com/bencandy.php?fid=119&id=204503

】【打印繁体】【投稿】【收藏】 【推荐】【举报】【评论】 【关闭】 【返回顶部
上一篇nginx ---->flume ----->ka.. 下一篇Windows64环境下   使用Flum..