这里通过介绍对于淘宝开放平台基础设置之一的TOPAnalyzer的代码优化，来谈一下对于海量数据处理的Java应用可以共享的一些细节设计（一个系统能够承受的处理量级别往往取决于细节，一个系统能够支持的业务形态往往取决于设计目标）。
先介绍一下整个TOPAnalyzer的背景，目标和初始设计，为后面的演变做一点铺垫。
开放平台从内部开放到正式对外开放，逐步从每天几千万的服务调用量发展到了上亿到现在的15亿，开放的服务也从几十个到了几百个，应用接入从几百个增加到了几十万个。此时，对于原始服务访问数据的分析需求就凸现出来：
1． 应用维度分析（应用的正常业务调用行为和异常调用行为分析）
2． 服务维度分析（服务RT,总量，成功失败率，业务错误及子错误等）
3． 平台维度分析（平台消耗时间，平台授权等业务统计分析，平台错误分析，平台系统健康指标分析等）
4． 业务维度分析（用户，应用，服务之间关系分析，应用归类分析，服务归类分析等）
上面只是一部分，从上面的需求来看需要一个系统能够灵活的运行期配置分析策略，对海量数据作即时分析，将接过用于告警，监控，业务分析。
下图是最原始的设计图，很简单，但还是有些想法在里面：
Master：管理任务（分析任务），合并结果（Reduce），输出结果（全量统计，增量片段统计）
Slave：Require Job + Do Job + Return
Result，随意加入，退出集群。
Job：(Input + Analysis Rule + Output)的定义。
几个设计点：
1. 后台系统任务分配：无负载分配算法，采用细化任务＋工作者按需自取＋粗暴简单任务重置策略。
2. Slave与Master采用单向通信，便于容量扩充和缩减。
3. Job自描述性，从任务数据来源，分析规则，结果输出都定义在任务中，使得Slave适用与各种分析任务，一个集群分析多种日志，多个集群共享Slave。
4. 数据存储无业务性（意味着存储的时候不定义任何业务含义），分析规则包含业务含义（在执行分析的时候告知不同列是什么含义，怎么统计和计算），优势在于可扩展，劣势在于全量扫描日志（无预先索引定义）。
5. 透明化整个集群运行状况，保证简单粗暴的方式下能够快速定位出节点问题或者任务问题。（虽然没有心跳，但是每个节点的工作都会输出信息，通过外部收集方式快速定位问题，防止集群为了监控耦合不利于扩展）
6. Master单点采用冷备方式解决。单点不可怕，可怕的是丢失现场和重启或重选Master周期长。因此采用分析数据和任务信息简单周期性外部存储的方式将现场保存与外部（信息尽量少，保证恢复时快速），另一方面采用外部系统通知方式修改Slave集群MasterIP，人工快速切换到冷备。
Master的生活轨迹：
Slave的生活轨迹：
有人会觉得这玩意儿简单，系统就是简单＋透明才会高效，往往就是因为系统复杂才会带来更多看似很高深的设计，最终无非是折腾了自己，苦了一线。废话不多说，背景介绍完了，开始讲具体的演变过程。
数据量：2千万 -->1亿 -->8亿 -->15亿。报表输出结果：10份配置
-->30份 -->60份 -->100份。统计后的数据量：10k -->10M
--> 9G。统计周期的要求：1天 -->5分钟 -->3分钟 -->1分半。
从上面这些数据可以知道从网络和磁盘IO，到内存，到CPU都会经历很大的考验，由于Master是纵向扩展的，因此优化Master成为每个数据跳动的必然要求。由于是用Java写的，因此内存对于整体分析的影响更加严重，GC的停顿直接可以使得系统挂掉（因为数据在不断流入内存）。
优化过程：
纵向系统的工作的分担：
从Master的生活轨迹可以看到，它负荷最大的一步就是要去负责Reduce，无论如何都需要交给一个单节点来完成所有的Reduce，但并不表示对于多个Slave的所有的Reduce都需要Master来做。有同学给过建议说让Master再去分配给不同的Slave去做Slave之间的Reduce，但一旦引入Master对Slave的通信和管理，这就回到了复杂的老路。因此这里用最简单的方式，一个机器可以部署多个Slave，一个Slave可以一次获取多个Job，执行完毕后本地合并再汇报给Master。（优势：Master在Job合并所产生的内存消耗可以减轻，因为这是统计，所以合并后数据量一定大幅下降，此时Master合并越少的Job数量，内存消耗越小），因此Slave的生活轨迹变化了一点：
流程中间数据优化：
这里举两个例子来说明对于处理中中间数据优化的意义。
在统计分析中往往会有对分析后的数据做再次处理的需求，例如一个API报表里面会有API访问总量，API访问成功数，同时会要有API的成功率，这个数据最早设计的时候和普通的MapReduce字段一样处理，计算和存储在每一行数据分析的时候都做，但其实这类数据只有在最后输出的时候才有统计和存储价值，因为这些数据都可以通过已有数据计算得到，而中间反复做计算在存储和计算上都是一种浪费，因此对于这种特殊的Lazy处理字段，中间不计算也不存储，在周期输出时做一次分析，降低了计算和存储的压力。
对于MapReduce中的Key存储的压缩。由于很多统计的Key是很多业务数据的组合，例如APPAPIUser的统计报表，它的Key就是三个字段的串联：taobao.user.get—12132342—fangweng，这时候大量的Key会占用内存，而Key的目的就是产生这个业务统计中的唯一标识，因此考虑这些API的名称等等是否可以替换成唯一的短内容就可以减少内存占用。过程中就不多说了，最后在分析器里面实现了两种策略：
1. 不可逆数字摘要采样。
有点类似与短连接转换的方式，对数据做Md5数字摘要，获得16个byte，然后根据压缩配置来采样16个byte部分，用可见字符定义出64进制来标识这些采样，最后形成较短的字符串。
由于Slave是数据分析者，因此用Slave的CPU来换Master的内存，将中间结果用不可逆的短字符串方式表示。弱点：当最后分析出来的数据量越大，采样md5后的数据越少，越容易产生冲突，导致统计不准确。
2. 提供需要压缩的业务数据列表。
业务方提供日志中需要替换的列定义及一组定义内容。简单来说，当日志某一列可以被枚举，那么就意味者这一列可以被简单的替换成短标识。例如配置APIName这列在分析生成key的时候可以被替换，并且提供了500多个api的名称文件载入到内存中，那么每次api在生成key的时候就会被替换掉名称组合在key中，大大缩短key。那为什么要提供这些api的名称呢？首先分析生成key在Slave，是分布式的，如果采用自学习的模式，势必要引入集中式唯一索引生成器，其次还要做好足够的并发控制，另一方面也会由并发控制带来性能损耗。这种模式虽然很原始，但不会影响统计结果的准确性，因此在分析器中被使用，这个列表会随着任务规则每次发送到Slave中，保证所有节点分析结果的一致性。
特殊化处理特殊的流程：
在Master的生活轨迹中可以看出，影响一轮输出时间和内存使用的包括分析合并数据结果，导出报表和导出中间结果。在数据上升到1亿的时候，Slave和Master之间数据通信以及Master的中间结果磁盘化的过程中都采用了压缩的方式来减少数据交互对IO缓冲的影响，但一直考虑是否还可以再压榨一点。首先导出中间结果的时候最初采用简单的Object序列化导出，从内存使用，外部数据大小，输出时间上来说都有不少的消耗，仔细看了一下中间结果是Map<String,Map<String,Obj>>，其实最后一个Obj无非只有两种类型Double和String，既然这样，序列化完全可以简单来作，因此直接很简单的实现了类似Json简化版的序列化，从序列化速度，内存占用减少上，外部磁盘存储都有了极大的提高，外部磁盘存储越小，所消耗的IO和过程中需要的临时内存都会下降，序列化速度加快，那么内存中的数据就会被尽快释放。总体上来说就是特殊化处理了可以特殊化对待的流程，提高了资源利用率。（同时中间结果在前期优化阶段的作用就是为了备份，因此不需要每个周期都做，当时做成可配置的周期值输出）
再接着来谈一下中间结果合并时候对于内存使用的优化。Master会从多个Slave得到多个Map<Key,Map<Key,Value>>，合并过程就是对多个Map将第一级Key相同的数据做整合，例如第一级Key的一个值是API访问总量，那么它对应的Map中就是不同的api名称和总量的统计，而多个Map直接合并就是将一级key（API访问总量）下的Map数据合并起来（同样的api总量相加最后保存一份）。最简单的做法就是多个Map<Key,Map<Key,Value>>递归的来合并，但如果要节省内存和计算可以有两个小改进，首先选择其中一个作为最终的结果集合（避免申请新空间，也避免轮询这个Map的数据），其次每一次递归时候，将合并后的后面的Map中数据移出（减少后续无用的循环对比，同时也节省空间）。看似小改动，但效果很不错。
再谈一下在输出结果时候的内存节省。在输出结果的时候，是基于内存中一份Map<Key,Map<Key,Value>>来构建的。其实将传统的MapReduce的KV结果如何转换成为传统的Report，只需要看看Sql中的Group设计，将多个KV通过Group
by key，就可以得到传统意义上的Key，Value，Value，Value。例如：KV可以是<apiName,apiTotalCount>,<apiName,apiResponse>,<apiName,apiFailCount>，如果Group
by apiName，那么就可以得到 apiName,apiTotalCount,apiResponse,apiFailCount的报表行结果。这种归总的方式可以类似填字游戏，因为我们结果是KV，所以这个填字游戏默认从列开始填写，遍历所有的KV以后就可以完整的得到一个大的矩阵并按照行输出，但代价是KV没有遍历完成以前，无法输出。因此考虑是否可以按照行来填写，然后每一行填写完毕之后直接输出，节省申请内存。按行填写最大的问题就是如何在对KV中已经处理过的数据打上标识，不要重复处理。（一种方式引入外部存储来标识这个值已经被处理过，因为这些KV不可以类似合并的时候删除，后续还会继续要用，另一种方式就是完全备份一份数据，合并完成后就删除），但本来就是为了节约内存的，引入更多的存储，就和目标有悖了。因此做了一个计算换存储的做法，例如填充时轮训的顺序为：K1V1，K2V2，K3V3，到K2V2遍历的时候，判断是否要处理当前这个数据，就只要判断这个K是否在K1里面出现过，而到K3V3遍历的时候，判断是否要处理，就轮询K1K2是否存在这个K，由于都是Map结构，因此这种查找的消耗很小，由此改为行填写，逐行输出。
最后再谈一下最重头的优化，合并调度及磁盘内存互换的优化
从Master的生活轨迹可以看到，原来的主线程负责检查外部分析数据结果状态，合并数据结果这个循环，考虑到最终合并后数据只有一个主干，因此采用单线程合并模式来运作，见下图：
这张图大致描述了一下处理流程，Slave随时都会将分析后的结果挂到结果缓冲队列上，然后主线程负责批量获取结果并且合并。虽然是批量获取，但是为了节省内存，也不能等待太久，因为每一点等待就意味着大量没有合并的数据将会存在与内存中，但合并的太频繁也会导致在合并过程中，新加入的结果会等待很久，导致内存吃紧。或许这个时候会考虑，为什么不直接用多线程来合并，的确，多线程合并并非不可行，但要考虑如何兼顾到主干合并的并发控制，因为多个线程不可能同时都合并到数据主干上，由此引入了下面的设计实现，半并行模式的合并：
从上图可以发现增加了两个角色：Merge Worker Thread
Pool和Branch merged ResultList，与上面设计的差别就在于主线程不再负责合并数据，而是批量的获取数据交给合并线程池来合并，而合并线程池中的工作者在合并的过程中会竞争主干合并锁，成功获得的就和主干合并，不成功的就将结果合并后放到分支合并队列上，等待下次合并时被主干合并或者分支合并获得再次合并。这样改进后，发现由于数据挂在队列没有得到及时处理产生的内存压力大大下降，同时也充分利用了多核，多线程榨干了多核的计算能力（线程池大小根据cpu核来设置的小一点，预留一点给GC用）。这种设计中还多了一些小的调优配置，例如是否允许被合并过的数据多次被再次合并（防止无畏的计算消耗），每次并行合并最小结果数是多少，等待堆积到最小结果数的最大时间等等。（有兴趣看代码）
至上面的优化为止，感觉合并这块已经被榨干了，但分析日志数据的增多，对及时性要求的加强，使得我又要重新审视是否还有能力继续榨出这个流程的水份。因此有了一个大胆的想法，磁盘换内存。因为在调度合并上已经找不到更多可以优化的点了，但是有一点还可以考虑，就是主干的那点数据是否要贯穿于整个合并周期，而且主干的数据随着增量分析不断增大（在最近这次优化的过程中也就是发现GC的频繁导致合并速度下降，合并速度下降导致内存中临时数据保存的时间久，反过来又影响GC，最后变成了恶性循环）。尽管觉得靠谱，但不测试不得而知。于是得到了以下的设计和实现：
这个流程发现和第二个流程就多了最后两个步骤，判断是否是最后的一次合并，如果是载入磁盘数据，然后合并，合并完后将主干输出到磁盘，清空主干内存。（此时发现导出中间结果原来不是每次必须的，但是这种模式下却成为每次必须的了）
这个改动的优势在什么地方？例如一个分析周期是2分钟，那么在2分钟内，主干庞大的数据被外置到磁盘，内存大量空闲，极大提高了当前时间片结果合并的效率（GC少了）。缺点是什么？会在每个周期产生两次磁盘大量的读写，但配合上优化过的中间结果载入载出（前面的私有序列化）会适当缓和。
由于线下无法模拟，就尝试着线上测试，发现GC减少，合并过程加速达到预期，但是每轮的磁盘和内存的换入换出由于也记入在一轮分析时间之内，每轮写出最大时候70m数据，需要消耗10多秒，甚至20秒，读入最大需要10s，这个时间如果算在要求一轮两分钟内，那也是不可接受的），重新审视是否有疏漏的细节。首先载入是否可以异步，如果可以异步，而不是在最后一轮才载入，那么就不会纳入到分析周期中，因此配置了一个可以调整的比例值，当任务完成到达或者超过这个比例值的时候，将开始并行载入数据，最后一轮等到异步载入后开始分析，发现果然可行，因此这个时间被排除在周期之外（虽然也带来了一点内存消耗）。然后再考虑输出是否可以异步，以前输出不可以异步的原因是这份数据是下一轮分析的主干，如果异步输出，下一轮数据开始处理，很难保证下一轮的第一个任务是否会引发数据修改，导致并发问题，所以一直锁定主干输出，直到完成再开始，但现在每次合并都是空主干开始的，因此输出完全可以异步，主干可以立刻清空，进入下一轮合并，只要在下一个周期开始载入主干前异步导出主干完成即可，这个时间是很长的，完全可以把控，因此输出也可以变成异步，不纳入分析周期。
至此完成了所有的优化，分析器高峰期的指标发生了改变：一轮分析从2分钟左右降低到了1分10秒，JVM的O区在合并过程中从50－80的占用率下降到20－60的占用率，GC次数明显大幅减少。
总结：
1. 利用可横向扩展的系统来分担纵向扩展系统的工作。
2. 流程中中间数据的优化处理。
3. 特殊化处理可以特殊处理的流程。
4. 从整体流程上考虑不同策略的消耗，提高整体处理能力。
5. 资源的快用快放，提高同一类资源利用率。
6. 不同阶段不同资源的互换，提高不同资源的利用率。
其实很多细节也许看了代码才会有更深的体会，分析器只是一个典型的消耗性案例，每一点改进都是在数据和业务驱动下不断的考验。例如纵向的Master也许真的有一天就到了它的极限，那么就交给Slave将数据产出到外部存储，交由其他系统或者另一个分析集群去做二次分析。对于海量数据的处理来说都需要经历初次筛选，再次分析，展示关联几个阶段，Java的应用摆脱不了内存约束带来对计算的影响，因此就要考虑好自己的顶在什么地方。但优化一定是全局的，例如磁盘换内存，磁盘带来的消耗在总体上来说还是可以接受的化，那么就可以被采纳（当然如果用上SSD效果估计会更好）。
最后还是想说的是，很多事情是简单做到复杂，复杂再回归到简单，对系统提出的挑战就是如何能够用最直接的方式简单的搞定，而不是做一个臃肿依赖庞大的系统，简单才看的清楚，看的清楚才有机会不断改进。 |
