大规模日志收集处理项目的技术总结

参考：http://sdjcw.iteye.com/blog/1814703

 

以下是2012年一个公司内部项目的技术总结，涉及到的方面比较多比较杂，拿出来和大家分享下。如果有更好的方案或者想法请联系我，谢谢～！
注：文章中提到的其他系统（如哈勃Agent、EagleEye）是公司内部的其他系统，这里就不详细介绍了。

简介

TLog是一个分布式的，可靠的，对大量数据进行收集、分析、展现的的系统。主要应用场景是收集大量的运行时日志，分析并结构化存储，提供数据查询和展现。

服务能力

• 收集淘宝线上8000+台机器的日志，每天日志量5T+。
• 一般情况下，数据从产生到TLog最终入库有10秒以下的延迟。
• 整个TLog集群有12台机器（虚拟机，5核，8G内存），10台负责日志的收集和解析，2台提供数据的查询和展现。

技术选型

一个海量数据收集的系统，首先需要考虑的就是收集模型：推送（push），还是拉取(pull)。两种模式都有各自的优缺点。业界的很多系统都是push模型，比如facebook的scribe，而我们主要选择的是pull模型（push模型后续支持），这个决策和我们所处的环境有关：

TLog集群可用资源非常有限

选用push模型，就需要要求日志收集器的容量需要大于高峰期数据的生成量，否则主动推送过来的数据不能及时处理会带来更多更复杂的问题：比如信息在收集器端如何先暂存慢慢处理，这又牵扯到收集器端是否有这么多的缓存空间（比如硬盘是否够大来临时保存汹涌而至的数据，或者转移到其他地方的网络开销等）；如果在日志生成端临时保存，则需要有一系列状态的变化，比如收集器正常则直接发数据，否则则保存本地硬盘，等到收集器恢复了再把硬盘数据发送，然后在恢复到直接发送模式等。

最初TLog集群只有6台虚拟机，后期扩展到12台。硬件处理能力的限制，决定了我们处理海量数据时压力非常大，如果还选用push模型，在数据生成的高峰期，必然无法处理瞬间大量的日志。而选择pull模型，控制权掌握在自己手里，收集器可以根据自己的节奏游刃有余的拉取日志，高峰期产生的日志会在接下来的时间慢慢的被消化（当然收集器的能力需要高于日志产生的平均值）。当然，这样的缺陷是处理延迟增加了。

信息的时效性

push模型能带来很高的信息时效性，可以最快的收集、整理，并查询出来。而我们的先期定位并不是特别在意这样的实时性，因为接入的应用主要是使用这些数据做日报、周报等，能够接受5~10分钟甚至更张的数据延迟。而且有些环境的约束导致做到秒级别的准实时也没有意义，比如HSF的哈勃日志，一个数据单元每2分钟才输出一次，从日志的输出端就已经造成了2分钟的延迟了，后面在快也意义不大。所以选用pull模型，在数据高峰期，大量数据临时挤压，后期慢慢处理对我们来说是可以接受的。

可靠性

可靠是必须的。众多push模型的产品在保证可靠性做了很多事情，使得事情变得非常的复杂，比如：

• 收集器出现异常：正常情况下直接推送消息，如果收集器异常则本地先缓存，待收集器恢复后再把缓存的信息发送，然后再恢复为正常模式
• 收集器的选择：当有多个收集器时，消息推送给谁？是否要负载均衡，是否要对收集器上下线很快的感知，势必需要引入ConfigServer或者ZK这样的产品
• 需要嵌入应用：如果要做到运行时信息不落盘直接发送，需要发送的逻辑嵌入应用，涉及到应用的大面积升级，而且上面提到了发送的逻辑在收集器的问题上变得不是那么简单，所以这对应用是很重的负担。

而选择pull模型，再借助哈勃Agent这个基础设施，事情会变得非常简单！这里不得不提一下：哈勃Agent是个很不错的产品，简单而有效！而且它的存在使得TLog设计和部署变得简单很多：

• 收集器无状态：这一点很重要，保证了TLog的可靠性和简单的设计。试想如果收集器需要记录每一个日志收集的状态（目标机器地址、日志文件、日志抓取偏移量），则收集器在重启或者挂掉时需要做很多状态持久化的工作，有时甚至没有时间来做而直接挂掉，所以又需要又更复杂的方式来近乎实时的记录状态；如果任务分配不均，或者其他原因向调动一些任务到其他的机器，又涉及到任务状态的迁移等等。而状态记录在哈勃Agent，收集器只需要通过相关标识请求哈勃Agent，即可获取增量日志，如果让任务调动或者某个收集器挂掉，别的收集器接替其工作，只需要使用同样的标识即可获取后续的增量日志，这样的结构使得对单个收集器的稳定性要求大大降低，只要整个集群持续有足够的资源，即可保证系统的可靠性。而且哈勃Agent足够的简单，使其很容易做到非常稳定。所以，不存在push模型的的问题一和问题二。
• 天然的分布式存储：开玩笑的说，TLog有个强大的分布式存储，即目标服务器本身的硬盘空间，每个服务器生成的日志直接落盘，而且硬盘空间又足够保持一定时间内的日志，收集器在之后的某个时刻读取这些信息，如果处理失败还可以根据上次日志抓取的偏移量再抓取一次或者保存再某处做特殊处理。这比所有日志都堆积到十几台收集器的硬盘上要可靠的多，也简单的多。
• 部署和依赖简单：几乎淘宝所有服务器都有部署哈勃Agent，即使没有也很方便的能安装。应用不需要嵌入发送日志信息的逻辑，只需要简单的记录日志即可，对应用几乎零侵入。所以，对push模型的问题三解决了。

当然，选择pull模型也是有自己的问题：

• 日志收集任务的管理：因为信息不会主动推送过来，所以需要自己记得去哪里取。收集任务的管理是个不小的问题：比如有些任务链接会失败（比如哈勃Agent没有部署，需要找PE解决）；有些任务会忙不过来，需要增加处理器节点（如某个任务的收集器负荷重导致收集频率降低，长时间没有抓取动作）；新的应用接入需要配置相关的抓取任务，应用的服务器变更后相关的任务也需要变更等，虽然很多都做了自动化处理（比如定时同步Armory来获取应用和ip的映射关系等），但不得不承认任务的管理是个不小的负担。

技术挑战

TLog做的事情非常简单，但是再海量数据的冲击下，系统很容易变得千疮百孔。

JVM内存溢出

TLog首先遇到的问题就是OOM。收集器所在的虚拟机，15MB/秒的数据流入10MB/秒的数据流出（这还是平常业务压力不大的时候）。很容易想象，10+MB的数据解析成大量的对象，稍微处理不好就会导致大量的JVM堆内存被占用，很容易OOM。结合应用自身的状况，经过很多尝试，最终找到了解决办法，这也让我对很多东西有了新的认识：

线程池的大小

线程池的大小对于TLog来说不是性能的问题，而是会不会死人的问题。线程池在TLog内部主要是任务调度使用（Quarz），每一个日志收集任务启动会占用一个线程，后续的所有动作都在这个线程完成：收集一批增量日志；使用不同的解析器把日志解析成结构化对象；持久化（入HBase或者云梯或者消息中间件）。这样的划分方式使得线程之间没有任何通信（也就没有锁的竞争），有因为整个处理任务的两头有大量的IO动作（拉取日志和持久化），中间过程是纯CPU运算（解析），所以多个线程大家互补忙闲能做到很高的效率（CPU和IO双忙……）。

但是线程池开多少？当初拍脑袋定了200，结果只要日志有积压（业务高峰，或者TLog下线一段时间）TLog直接OOM。中间甚至使用过“延迟启动任务”的方式，即收集器把任务以一定间隔（比如2秒）一个一个启动，有一定效果，但还是很容易挂，而且一个收集器一般会有5k+个任务，两秒启动一个的话……这很显然不靠谱。分析了状况后，发现事情是这样的：

1. 收集器开启一个任务抓取日志，返回的量很大（比如10MB），拿到内存里开始解析，解析完开始保存。
2. 于此同时，其他的任务也启动起来，做同样的动作，拉日志，解析，保存。
3. 因为线程池有200个线程，意味着同时允许200个任务在运行，而我后续dump内存发现，一个线程运行的时候，会占用20~40MB不等的JVM内存。而总共JVM堆有5G，还没到200个线程运行内存就不够了，然后开始GC，但是GC掉的内存不多，因为很多线程刚刚处于数据保存阶段，数据在最终入库前是不可能GC的，所以港GC完内存又瞬间用光然后再次GC，或者就干脆OOM了。

原本很简单的事情（拉日志，解析，入库）变得无法稳定运行，经过一步步测试，最终把线程池大小控制在30（后续因为逻辑更加复杂，但任务占用内存量又增加，而调整到25），之所以调整到这么小是因为如果再大，比如35个任务同时处理，就会导致内存占用非常紧张（虽说有5G的堆内存，old、eden、s1和s0分分就没多少了），导致Full GC，但又没有成果，GC完内存一样不够用，就再Full GC，结果导致90%以上的时间都在做无用的GC。那还不如把内存控制的留点余量，不至于频繁触发FGC，而留下大量的时间专心干活呢。当然除此之外还有很多其他的优化，比如把先批量解析再批量保存改为边解析边保存，保存过后的对象就可以被GC了，降低对象的存活时间。另外一个很重要的点：通过哈勃Agent拉取的一批增量日志一下都被加载到内存中，随后慢慢的解析处理。在极端情况下只要这批日志没有处理完，就会有10MB（哈勃Agent单次拉取日志的上限）的字节无法释放。应该改为流式的处理，读一部分处理一部分，然后再读取下一部分。但因为这个改动对整个解析结构会有很大的调整，所以就放到了后面迁移Storm时统一做了修改，整个jvm内存占用量减少一半左右，不再成为系统的瓶颈。

经过了一番改动后，TLog变得“压不死”了，即使积压了大量的数据，启动后网络流量涌入30MB/每秒，cpu会稳定再85%，系统load 40（夸张的时候有80），但是很稳定，不到2秒一次YGC，10分钟一次FGC，系统可以保持这样的压力运行6个小时，当积压的所有数据都处理完后，机器的负载，cpu及内存的占用自动恢复到正常水平。

JVM GC参数

最初TLog使用的GC方式是CMS GC，花了不少经历调整eden和old的比例，以及触发CMS GC的比率。但后来觉得这样做没有必要。因为CMS GC是为高相应系统设计的，使Stop-the-world时间尽量短，使得系统持续保持较高的相应速度，但付出的代价就是GC效率低。而TLog选用的是pull模型，不会有系统主动请求，所以不需要保证高相应，应该更加看中GC效率，所以后续改成了并行GC，提高GC效率，从而获得更多的“工作时间”。

另外eden和old的比例对收集器也非常重要，从工作方式可以看到，收集器必定会大量的产生对象，和大量的销毁对象，而且这些对象还是会在内存中保留一定的时间，所以要求eden区稍微大些，以保证这些临时对象在晋升到old之前就被回收掉，而相对稳定的数据在收集器中比较少。所以eden空间设置的比old要大。

HBase写效率

随着业务量的增长，马上遇到了一个之前从没有想象过的问题：HBase写速度不够。当初之所以选择HBase为后端存储，就是因为其写入速度很高，能保证大量的数据快速的入库。先前的测试也验证了这一点：HBase单机扛住了我们每秒5万条记录的冲击。所以我们乐观的估计，有着30台机器的集群抗100万的量应该小case，但是随着越来越多的应用使用HBase集群，以及整个集群数据量的增加及region数量的增多，HBase写效率不断下降，同样保存1k条记录的耗时从原来的不到100毫秒变成了将近1秒，有时甚至超过2秒。直到一个周六的晚上，整个集群写入耗时忽然急剧上升，导致整个集群所有应用的写入量被迫下降到10万/秒左右，最后无奈关闭了Eagleeye的数据表，整个集群才恢复，原因很简单：Eagleeye数据表的region数量将近1万个，占整个集群region数量的80%，region server压力过大。至此Eagleeye实时数据就暂停下来，全部转为离线处理。

Eagleeye是TLog最大的一个接入方，其数据量占TLog所有业务的80%，每天日志量5T左右。HBase上的数据表被关闭，一部分原因是数据量的确太大，另外我觉得应该是我们使用HBase的方式不够得当，还有优化的空间。所以我开始寻找业界的解决方案，发现了OpenTSDB。

OpenTSDB

"OpenTSDB is a distributed, scalable Time Series Database (TSDB) written on top of HBase."（官方说明）。学习了OpenTSDB的Schema设计，发现很多东西都值得学习和借鉴，根据多面对的场景，OpenTSDB对Schema的精巧设计，使得其记录体积非常小，而且row的数量很少，这都能降低HBase和region server的压力，从而提高数据库的写和读的效率。将HSF的数据改为OpenTSDB的方式后，同样的信息量，数据体积减少了80%以上（rowkey体积减少50%，value体积减少80%，数据条数减少66%），rowkey数量减少97%。直接效果就是写入速度更快，吞吐量更高，而且HBase服务器的压力更小！但这只适用于Time Series类型的数据，比如HSF、精卫等数值统计型的场景。对于Eagleeye和TAE这种日志记录类型的不适用，但仍然又很多可以借鉴和改进之处。

流式处理的代价

下半年，TLog的收集器迁移到Storm流式处理平台。日志收集器兼顾了收集、解析、入库的职责，而且解析期间经常需要对信息进行分类，过滤，汇总等，非常适合使用流式处理框架完成这些工作。但迁移到Storm后一样遇到了各式各样的挑战：

额外的消耗

在Storm中，每个处理节点可以认为是一个运算单元，数据在这些单元中流转，一级的输出作为另一级的输入。对于TLog的解析器来说，感觉理想的方式应该是这样的：

1. Spout节点（Storm中的数据源节点）拉取日志内容，然后发射出去
2. 下游节点接收日志内容，进行解纷并结构化成对象，然后发射出去
   1. 下游节点接收结构化对象，进行相关存储（HBase或者云梯）
   2. 另一类下游节点接收结构化对象进行相关的聚合，比如根据某种类型进行累加

上面的处理方式感觉非常清晰明了，但是却产生了大量的“额外消耗------对象的序列化和网络传输。数据在每个节点流转都需要经过序列化和反序列化操作（消耗CPU），还有网络传输（消耗IO），而且根据上面的设计，几乎从spout流出的数据会100%的跳转多个节点，也就使得一份数据造成N倍的网络传输，网络消耗非常严重。所以我们制定了一个简单的原则：只要没有聚合需求，就在一个节点完成。因为集群数据的聚合使用普通方式比较难解决，而使用storm非常天然的处理掉。

对避免不了的数据流转，storm还是有办法降低额外的消耗，比如：

• 优化序列化方式：storm可以使用kryo的序列化方式，cpu消耗和序列化后的体积会比java自身的序列化好很多，可以参考这里的比较。
• 进程内流转不做序列化和网络传输处理：storm 0.7.2版本做了优化，如果数据是在一个进程内流转，则跳过序列化和网络传输步骤，这样能极大的减轻额外的消耗。但是这需要使得多个节点在一个进程，会使得进程的庞大，导致机器storm worker进程数减少，可能造成负载不均衡的情况，所以一台机器开多少个worker需要根据机器的配置，以及任务的复杂度，以及任务数量来权衡。

可靠消息和非可靠消息的选择

Storm为了保证流转消息的可靠性，引入第三视角的节点Acker，来跟踪每一条消息，当下游处理失败后能通知上游，上游可以有自己的策略进行处理（例如重发消息）。但是Acker的引入也必然有开销（大量的Ack消息），导致业务可用的资源减少，而且会降低消息处理的性能。TLog处理器未启用可靠消息时，每个节点处理消息的速度是11k/s，打开可靠消息后只能有3~4k/s，下降非常明显。因为TLog处理的是大量的日志信息，处于从数据可靠的敏感程度，和资源限制的情况下，我们选择了非可靠消息。

但事情并没有这样简单的结束，我们的集群经常出现个别进程内存狂涨，消耗掉所有的内存甚至swap分区，然后操作系统启动自我保护性的随机kill进程，导致这个“异常”进程被杀死；或者整个虚拟机挂掉。出现这个问题的原因是我们生产消息（Strom的spout端）的速度大于消费消息（Strom的bolt端）的速度，导致消息积压在spout端的出口处，使得spout所在的进程内存占用上升（顺便提一下，Storm使用的消息组件是ØMQ，非java组件，所以消息的堆积无法从jvm堆体积中体现出来）。而Storm可以通过设置“topology.max.spout.pending”来设置积压消息的最大值，但是这个特性只有在“可靠消息”时才有意义。所以对于非可靠消息，只能提高后续节点的处理能力（比如增加节点数量）来解决。

实时和离线相结合

对于运行时数据，一般情况下我们的场景如下：

• 近期的实时数据：对于这部分数据，我们的需求是查询时间跨度小（近1小时或近10分钟），时间粒度细（每分钟甚至几十秒一个数据单元），能够准实时的展现（数据从产生到最终展现的延迟可能只有几秒或十几秒）。
• 过往的历史数据：对于历史数据，我们的查询的时间跨度一般比较大（一周、一个月等），但时间粒度较粗（一小时甚至一天一个数据单元），不在乎实时性。

所以对数据粒度的需求会随着时间的流式而变粗（ps：你应该不会需要查看上个月3号上午10点～10点半，以分钟为粒度展现一个服务的调用量。如果真的需要，这应该是一个特殊情况，相关的报警系统应该会沉淀该信息）。所以从“如何“打败”CAP定理”一文得到的思路，我们使用实时和离线相结合的方法来解决一下需求：

实时部分

准实时的处理最新的数据，以小粒度保存（甚至可以直接缓存起来），方便查询和检索。但实时处理数据有一些问题：

• 易错：实时处理一般使用流式处理，大量数据批量涌入又快速的处理输出，很容易出现不确定性或错误。如数据因为收集的不同步，导致加和时上一分钟的日志被加和到下一分钟；或者因为短暂的暂停服务导致数据出现缺口；或者一个新上的算法有缺陷导致计算错误等。
• 无法重复计算：因为数据快速的流动，如果有消息重发机制就意味着一定有个池子来缓存，以保证下游处理失败而重发。缓存池的存在又增加了资源、性能、复杂度等的极大提高。
• 数据量太大：因为数据的时间粒度太细，使得数据量非常大，存储和查询代价很高。

所以我们的做法是在实时部分允许有这样细小的问题，问题的修复由离线批量计算解决。

离线批量处理部分

使用MapReduce来计算一段时间（前一小时或一天）汇总的数据，很容易解决实时计算是出现的问题：

• 大时间跨度的合并：单条数据汇总时间跨度较大，极大减小了数据的体积，对存储和大时间跨度查询友好。
• 替代“过时”的实时数据：实时计算的结果会被批量计算覆盖或替代，当时产生的细小错误也自然消失。
• 容错性高：因为原始数据已经保存，使用MR可以重复计算，而且计算结果稳定。即使出现算法错误，修复后仍然可以重新计算。

而离线批量处理的唯一问题-----实时性-----也被实时计算弥补。

数据说话

前面提到了很多系统优化和调整的方式，但一定要记得“要进行优化，先得找到性能瓶颈！”，根据Profiler的结果来确定优化的方向。对于TLog收集器，使用了Java自带的VisualVM，很快定位到几个最大的cpu消耗点：

• 反射构造对象：为了方便构造存储HBase的结构化对象，我们开发了一套注解，通过在Model对象属性上标记注解，可以自动转换成需要的HBase对象，使用起来非常方便，但是转换过程没有缓存Model的类结构，导致大量的使用反射。这样生成对象的速度比直接代码构造要慢了一个数量级。
• 正则表达式匹配：收集器中有大量的日志匹配和解析，所以当时用了很多正则表达式，虽然缓存了pattern，但是还是非常的消耗cpu，最后把一些非常规范的日志都使用StringUtils.split()方法进行分割然后处理，也能提大幅度提高解析速度。
• 切分字符串：Java String的split方法内部使用了曾则表达式，如果切分字符不需要正则匹配，建议使用Apache commons lang的StringUtils。当然，这个的调整带来的提高远不如前两个大。

经过一次次Profile和调整，最终基本只剩下无法避开的消耗，处理器容量提高4倍以上！遗憾的是当时前后的详细对比数据没有保留，无法列在这里提供参考。

其他

下面是一些零碎的小心得。

HBase rowkey的唯一

TLog收集到的一些信息不是Time Series类型的，不能做加和等处理，而是要根据日志内容生成唯一的个体，比如操作日志，需要能根据时间和操作类型查询操作的具体情况。对于这类需求有个细节：HBase rowkey的生成该如何保证唯一。因为rowkey会由索引条件构成，如日志类型、时间，但仅仅这样的rowkey很容易重复，导致之前的记录被覆盖。当然可以在rowkey后面增加一个唯一后缀进行区分，比如下面几种方式：

• 增加相关区别标识：比如增加日志生成机器ip，或者其他什么比较容易区别的业务字段信息来进行去重，但是会发现后缀元素加的少很难达到效果，加的多又使得rowkey变得很庞大，甚至只有把所有日志内容都放在rowkey里才能达到去重目的。
• 增加递增量：增加一个自增的变量，但是这样的缺陷是数据无法补救，即如果因为某种原因，昨天一天的数据想重新导入一边，昨天已有的覆盖掉，没有的补上，你会发现自增量的存在导致你没办法做这个事情。

我们的处理方式：将日志原文进行CRC-32编码，生成8位16进制的值，附加在rowkey的末尾，即保证了rowkey不会过度的膨胀（最多8个字符的长度），又保证了低重复率（CRC-32碰撞几率相对较低），而且可以支持数据的重复导入（相同记录计算的编码一样）。