并没有完整解决，依靠flume来缓解集群的贷款压力实现部分实时

使用Flume做日志收集，flume+kafka做日志收集存储，后面接es存历史日志，storm用于实时分析。对于历史日志来说，尽可能做到不丢不重，实际经验内网环境kafka还是很稳定可靠的，丢日志主要发生在flume重启，所以针对flume进行改造，实现断点续传。主要环境及配置，jdk7，flume1.6版本，agent使用内存channel。自扩展source基于原生ExecSource改造，sink由KafkaSink改造。背景：我们日志文件名格式有2种，一种是按时间段分隔，比如文件名为debug.log，debug.log.2016-05-18-15，debug.log.2016-05-18-14，...；另一种为msg.log，msg.log.1，msg.log.2，...

设计思路

设计思路：

• 1. 在sink往kafka发送完后，记录原始日志文件及行号信息到record文件（记录文件内容只有几个字节，内容为：debug.log,2016-05-18-15,265。第二个字段为时间槽或者文件的最后修改的long型毫秒时间）
• 2. 每次重启时，在source中检查record文件是否存在，存在的话读取上次最后一条日志的文件名和行号
• 3. 如果第2步的record文件记录和当前最新的需要抽取日志文件是同一个，且行数差距在1000以内（具体可以配置），直接tail -f -n +${上次抽完的行号+1} --pid ${flume的进程号} debug.log
• 4. 如果第2步的record文件记录和当前最新的需要抽取日志文件是同一个且行数差距在1000以上；或者record文件记录里原始日志文件名和当前的不是同一个，直接tail -f -n +${文件的最新行号+1} debug.log；另外记录这段落下的日志信息到快照文件snapshot，快照文件内容为：debug.log,2016-05-18-14,5432,debug.log,2016-05-18-15,1234
• 5. 在source中启动tail命令后，检查是否有快照文件snapshot，如果有则开启一个线程把快照文件里记录的日志读取直接写到kafka。

部分细节

• 1. record文件的写入，采用ByteBuffer，避免频繁IO，在指定时间间隔后flush到磁盘
• 2. 重启过程中，不要使用kill -9结束flume，使用kill -15，在接收到停止信号时，所有的source和sink快速停止操作，等待flume框架来调用每个source和sink的stop方法，在stop方法中刷新ByteBuffer内容到磁盘。
• 3. 为什么要在sink里记录行号, 主要是保证记录往kafka里写入成功的行号，另外不用tail -F 也是为了在日志文件名切换时重新从1开始计数行号
• 4. 采用fileChannel的，相应的在source写入channel成功后，记录行号即可

实际测试结果，在一个每秒产生300条日志的单个日志文件，做flume重启，没有丢失数据。

同时，可以看下这篇文档：

https://baijiahao.baidu.com/po/feed/sharewfr=spider&for=pc&context={"sourceFrom":"bjh","nid":"news_3433179683779105534"}

多看几篇文章，搞清楚，flume和断点续传的原理和意义。

那接下来，是不是得咱们搭建开发环境了？

下载源码很简单，去官网或者去github下载，Apache Flume 1.7.0的github源码地址如下：

https://github.com/apache/flume/tree/release-1.7.0

在mvn install之前，最好先设置下maven的国内镜像地址，加快依赖的下载速度，时间还是很宝贵的，别浪费在无聊的等待上，

打开maven的setting.xml配置文件，添加如下镜像即可：

<mirrors>

<mirror>

<id>alimaven</id>

<name>aliyun maven</name>

<url>http://maven.aliyun.com/nexus/content/groups/public/</url>

<mirrorOf>central</mirrorOf>

</mirror>

</mirrors>

搭建好了，从左边开始，依次依赖，我们要修改的taildir-source是flume1.7引入的，断点续传，针对文件系统的，但是hdfs的还得看咱们做。

然后我们想想办法能不能测试。