flink 运维系列之任务日志收集方案

flink任务日志指的是任务系统日志与用户代码里面log方式打印的日志，这些日志信息都可以在flink web页面上看到，目前任务的部署模式都是on yarn, 那么在yarn页面也可以看到，这些日志信息在开发环境或者测试环境量都是很小的，可以很方便的查看，但是在产生环境上，任务是7*24不间断的运行，那么势必会造成日志量会很大，这时打开flink web页面查看任务日志信息就会造成浏览器卡死，很难通过日志排查问题，所以需要将日志发送到外部的搜索系统中，方便搜索日志。

关于开源的日志收集方案，很可能想到的就是Elasticsearch+Logstash+Kibana，也就是我们常说的ELK，通过Logstah将日志收集起来发送到Es中，然后通过Kibana查询Es中的数据，那么在这里主要需要考虑的就是Logstash如何搜集flink任务日志，需要在每台nodemanager节点上安装Logstash去收集日志，由于任务可能会重启那么日志的目录也是一个动态变换的，这种方式势必会占用集群的资源，同时后期集群扩容也需要在新的节点上安装Logstah，由于这些因素使用一种新的方案：通过log4j的方式将日志发送到kafka，logstash消费kafka 的数据。

flink 任务日志打印提供了log4j 或者logback的方式，这里选择比较熟悉的log4j，log4j日志发送到kafka有成熟的插件KafkaLog4jAppender可以使用，但是log4j默认打印的日志并不是es能够解析的json格式的数据，另外flink任务日志中并不包含任务所对应的applicationId信息，在kibana查看起来会比较麻烦，所以采用自己log4j Appender方式，将数据发送到kafka, 并且定义符合规范的数据格式同时获取到任务的applicationId。在flink 目录的conf下面，有一个log4j.properties文件，负责任务日志相关配置，在默认情况下会有一个{log.file} 变量， 这个变量是taskManager启动传入的参数，表示的是日志路径，路径里面包含了applicationId, 那我们可以将{log.file}传给我们自定义的appender, 然后将applicationId 解析出来，接下来看下实际代码：

继承AppenderSkeleton 类， 里面定义了kafka 相关参数、logFile参数，日志打印发生在append 方法， 日志写入kafka的成功与否不应该影响flink任务的执行，所以这里做了try/catch并且异步发送kafka，在AppendKafkaTask 里面执行具体的数据格式化、applicationId解析 发送至kafka里面。

log4j.properties 配置：

然后将自定义的log4j appender 打包，放在flink 的lib目录下面，任务启动就可以在kibana查看到日志信息。