由定时脚本错误以及Elasticsearch配置错误引发的Flink线上事故

近期接手离职同事项目,突然遇到线上事故,Flink无法正常聚合数据生成指标.

以下是详细的排查过程:

问题复现

清晨,运维报告Flink数据分析模块无法正常生成指标数据.

赶紧登陆Flink所在机器,使用如下语句简单查看Job状态.

./bin/flink list

查看输出,发现故障JobRunning状态.

因为数据分析模块运行时间较久,近期没有更新过,因此怀疑是依赖的中间件问题.

问题根源定位

(1) 查看数据源

数据分析模块依赖于Kafka,因此登陆Kafka所在机器,查看相应topic的消费情况.

./bin/kafka-consumer-groups.sh  --describe --bootstrap-server 192.*.*.*:9092 --group data_prod_consumer

得到结果如下所示:

TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
qq_data 0 340155122 340155407 285 - - -

多次运行命令(间隔一段时间),发现LOG-END-OFFSET的增长速度远大于CURRENT-OFFSET,导致LAG指标不断上升,最高时达到17000.

让同事紧急修改日志输出,屏蔽非核心业务的日志,然而作用不大,LAG仍然在缓慢的持续增长.

LOG-END-OFFSET意为当前最新数据偏移量,CURRENT-OFFSET意为当前消费者消费的偏移量,LAGLOG-END-OFFSETCURRENT-OFFSET之间的差值.

通过LAG数值,可以观察到topic的消费情况,从而确定是否有数据积累.

观察flink机器负载,发现负载较低,消费瓶颈不应该是Flink.

(2) 查看数据输出

数据分析模块的SinkElasticsearch(单节点运行),因此查看ES的运行状况.

curl 'localhost:9200/_cat/indices?v'

查数据分析模块输出的数据库,发现数据库大小正常,感觉疑惑.

使用top查看机器的负载信息,大吃一惊:

由定时脚本错误以及Elasticsearch配置错误引发的Flink线上事故

由上图可知,ES应用的虚拟内存(VIRY)占用高达28G,此外,机器的load average高达4.43.

基本上,可以确定是Elasticsearch异常引发数据分析模块运行异常.

重新查看ES索引信息,发现有四个Type数据量不对(非相关业务数据表),其中一张表的数据量多达2800万条(此业务有过了解).

询问离职同事,经过询问大致可确定这四张表的数据量不对,此表数据未设置TTL,而是依赖于数据库定时删除脚本完成过期数据删除操作.

(3) 确定问题

查看数据库定期删除脚本执行日志,发现脚本在较长的一段时间内未能正常运行.

原因是脚本在这执行删除操作时,未根据业务的变更而修改脚本,导致在删除已废弃表中数据时出现异常而终止运行.

因为当前故障业务的数据库删除命令在异常前,故当前业务的数据量维持在正常的水平线上.

ps -ef|grep elasticsearch
/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.161-2.b14.el7.x86_64/jre/bin/java -Xms1g -Xmx1g -XX:+UseConcMarkSweepGC -XX:CMSInitiatingOccupancyFraction=75 -XX:+UseCMSInitiatingOccupancyOnly -XX:+AlwaysPreTouch -server -Xss1m -Djava.awt.headless=true -Dfile.encoding=UTF-8 -Djna.nosys=true -XX:-OmitStackTraceInFastThrow -Dio.netty.noUnsafe=true -Dio.netty.noKeySetOptimization=true -Dio.netty.recycler.maxCapacityPerThread=0 -Dlog4j.shutdownHookEnabled=false -Dlog4j2.disable.jmx=true -XX:+HeapDumpOnOutOfMemoryError -Des.path.home=/home/app/elasticsearch-6.1.2 -Des.path.conf=/home/app/elasticsearch-6.1.2/config -cp /home/app/elasticsearch-6.1.2/lib/* org.elasticsearch.bootstrap.Elasticsearch -d

此外,查看Elasticsearch的运行信息时,发现其配置明显存在不足,Xms分配为1G,Xmx分配为1G.

然而当前机器仅运行Elasticsearch,因此暂停依赖于Elasticsearch的相应业务,调整配置信息,重新启动.

ps -ef|grep elasticsearch
/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.161-2.b14.el7.x86_64/jre/bin/java -Xms8g -Xmx8g -XX:+UseG1GC -XX:MaxGCPauseMillis=50 -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+ParallelRefProcEnabled -XX:+AlwaysPreTouch -server -Xss1m -Djava.awt.headless=true -Dfile.encoding=UTF-8 -Djna.nosys=true -XX:-OmitStackTraceInFastThrow -Dio.netty.noUnsafe=true -Dio.netty.noKeySetOptimization=true -Dio.netty.recycler.maxCapacityPerThread=0 -Dlog4j.shutdownHookEnabled=false -Dlog4j2.disable.jmx=true -XX:+HeapDumpOnOutOfMemoryError -Des.path.home=/home/app/elasticsearch-6.1.2 -Des.path.conf=/home/app/elasticsearch-6.1.2/config -cp /home/app/elasticsearch-6.1.2/lib/* org.elasticsearch.bootstrap.Elasticsearch -d

修改后,Xms以及Xmx响应调整为8G,按照Elasticsearch推荐配置,保留8G内存供系统使用.

此外,临时修改数据库脚本并执行,删除各表中的过期数据.

curl -XPOST "http://localhost:9200/data/_delete_by_query?conflicts=proceed" --header 'Content-Type:application/json'  -d @delete_data_condition.txt

使用上述语句删除过期数据,删除条件置于delete_data_condition.txt文件中.

需要注意,因为Elasticsearch版本控制的原因,容易导致数据删除失败,因此添加conflicts=proceed条件解决版本冲突导致的数据删除失败.

问题解决

经过上述操作后,发现Elasticsearch数据库的负载明显下降,观察Topic的消费情况,发现LAG指标在快速回落.

待观察一段时间后,发现数据分析模块能够正常聚合,业务恢复正常.

不过还是留下了问题,就是因为Elasticsearch导致Flink消费Kafka数据过慢,出现了重复读进一步拖慢聚合速度,这个问题需要后续深入研究解决.

这就是一个删除脚本异常引起的血案,中间件虽然带来了极大的便利,但是也提升了问题排查的难度.

因此,不仅要知道中间件怎样部署,更要学会如何部署更优,为业务的正常运行奠定基础.

PS:

如果您觉得我的文章对您有帮助,请关注我的微信公众号,谢谢!

由定时脚本错误以及Elasticsearch配置错误引发的Flink线上事故

上一篇:java链接并操作数据库


下一篇:Atitit 图像处理 调用opencv 通过java  api   attilax总结