作者 阎志涛 [TOC] ## 前言 随着移动互联网时代的到来，越来越多的与人、与物、与环境有关的数据产生，大数据技术也变得越来越重要。在国内，大数据也由几年前的概念阶段逐渐的在不同的企业和行业落地，并且对企业的运营、发展起到了越来越重要的作用。从2011年创业之初，TalkingData就坚信一句话：”In God we trust, everyone else must bring data”. 期望通过数据去改变人们做决策的方式，通过数据让人们更好的了解自己。创业这4年，对于TalkingData技术团队来讲，也是对大数据技术架构的认识逐步深入的过程。经过4年的发展，我们每天处理的新增数据由几个GB逐渐的增加到如今的数个TB。而数据计算类型也从最初的统计分析的类型到支持多维交叉、即席查询、机器学习、广告归因等等多种计算类型。一路跌跌撞撞走来，TalkingData技术团队在大数据技术上踩过不少坑，有过许多个不眠之夜，也逐渐总结了一些自己的经验，并在这里分享给大家。 ## 大数据平台之初试 提到大数据平台，首先需要考虑的是采用什么样的技术架构。TalkingData在成立之初，主要业务是移动统计分析业务，主要帮助我们的客户分析移动应用的新增、活跃、留存、流失等等指标。包括这些指标在渠道、版本等不同维度的分布。在2011年，大部分的分析系统还都是纯粹离线的分析系统，所有的指标都是T+1才能获取。但是考虑到用户的体验，所有的数据都是T+1显然是不能够满足用户的需求的，用户需要知道当天的实时数据的分析结果。为了解决这个问题，大数据平台的高层架构如下：  数据通过数据收集器进行接收，接收后的数据会发送到消息队列中。ETL负责将数据进行规范化和逻辑组织、抽取，然后发送给实时计算部分和离线批量计算部分。实时计算部分主要负责计算当天的实时数据，而离线批量处理部分则主要计算当天以前的数据。查询API则提供统计分析业务对计算结果的查询。 具体到采用的技术，平台的组件架构图如下：  在这个架构当中，不同的组件采用的技术分别为： ## Data Collector（收据收集器） Data Collector主要负责收集从SDK发送来的各种数据，以日志的形式保留在本地，然后再将数据发送到消息队列中。 整个Data Collector组件包括Nginx作为负载均衡器，接收所有从SDK发送来的请求数据并发送到后方的真正处理数据的Data Collector中。Data Collector是运行在Jetty容器中的java servlet，利用容器提供的多线程的支持，接收并处理SDK发送上来的数据。数据会先以日志的形式存储在本地磁盘，然后再将数据发送到Message Queue中。Data Collector由于承担了数据接收的工作，在设计实现中不承担任何的计算逻辑，主要承担的是存储和转发的逻辑，从而能够高效的接收数据。 ## Message Queue（消息队列） 消息队列主要是为了解耦数据接收和数据计算的逻辑，在第一个版本中，采用的是轻量级的消息队列Kestrel。作为一个轻量级的消息队列，Kestrel非常的轻巧和方便使用，并且能够支持消息的存储，对消息的访问支持memcached协议，并且有非常不错的读写性能。对于快速构建一个支持异步处理的分布式系统来讲，Kestrel无疑是一个非常简单方便的选择。 ## Batch Computing（离线处理） Batch Computing主要是对非实时要求的数据做批量处理，在2011年，想做离线批量计算，能够选择必然是hadoop生态系统中的某种技术。可以选择自己写MapReduce，也可以选择Pig或者Hive来完成对应的工作。考虑到开发的方便性，Hive因为其支持类似于标准SQL的HQL最终被我们选择为离线处理的计算平台。批量计算的结果，会存储到Result Store中。为了解决多维交叉的问题，在批量处理过程中，我们会对每个维度生成对应设备的bitmap索引，同时也会将索引存储到Result Store中。 ## Real-time Computing（实时计算） Real-time Computing主要是为了解决客户需要看到实时分析数据的结果需求而引入的组件。在离线处理部分，我们通过Hive来计算一天以前的数据，包括各种时间跨度比较长的指标。不过对于用户当天当时的各种指标，Hadoop生态系统中的各种技术，因为其设计就是为离线计算而生，就不能够满足实时计算的要求了。在2011年，还没有非常好的开源的实时处理框架。我们能够选择的只能是自己去根据业务的需求开发自己的实时计算的组件。整个实时计算组件是采用Redis内存数据库为基础实现的。利用Redis提供的高速的访问能力，以及对能够对key的值进行增加计数，可以设置key过期等能力，我们将实时的计算指标通过组织Redis key来完成。 ## Qeury API（查询接口） 数据在离线计算和实时计算完成之后，会被Query API进行查询。Qeury API会分别从Redis当中查询当日的实时结果，同时会从Result DB查询离线计算的结果。为了方便查询，我们将Query API进行了封装，封装成了一个支持类似于SQL的查询引擎。分析业务会通过类SQL的表达将查询发送给查询API，查询API会根据时间切片的不同，决定是从Redis还是从Result DB中查询数据，并且将结果拼装后返回给分析业务系统应用。现在回头看我们的架构，实际上像极了后来Storm的开发者Nathan Marz提出的Lambda架构，其架构如下：  从上图可以看到，我们在2011年采用的架构和Lambda架构非常的像，只是采用的技术实现不尽相同。 ## 大数据平台之改良 随着业务的发展和数据的增加，2011年我们那套技术架构也逐渐出现各种各样的问题，对架构的重构也就变得越来越重要。这套架构主要存在的问题主要包括： ## 数据一致性问题 由于数据计算存在实时和离线两个部分，实时计算采用的Redis，而离线部分采用的是Hive，由于SDK上传的数据可能出现延迟，这样实时计算的时候可能当天没有上传的数据，在以后会上传，这样实时计算的当天的数据和当天过后通过Hive进行批量计算的结果就会产生偏差，有时会造成用户的困惑。 ## 数据处理能力不足问题 随着数据量的逐渐增加，这套技术架构也开始面临数据处理能力的考验。Kestrel作为一个轻量级的队列，能够满足初期的要求。可是数据量增多后，kestrel的平行扩展能力的不足开始体现，另外随着数据业务的增加，消息队列模型需要更灵活的支持多消费者的消息队列。而kestrel在这方面也很难满足业务的需求。 为了解决这些问题，我们决定对架构进行重构，于是一套自行研发的计算框架Torch（火炬）系统应运而生。这套系统采用微批次的概念，主要解决大数据场景下统计分析业务的需求。整体的技术架构如下：  这个架构中，大家可以看到，消息队列从Kestrel变为Kafka。采用Kafka，在数据量每天都在增加的时候，更方便的进行平行扩展。另外，业务可靠性的要求也越来越高，而Kafka本身的高可靠性的特点也更适合业务的需求。 在这个改良的架构中，不再存在实时和离线处理两个数据计算路径，所有的数据计算都是通过Torch的计算引擎来完成。整个计算引擎分为Counter引擎和Bitmap引擎两个部分。计算是以分钟为单位的微批次的计算，Counter引擎主要进行汇总类型的计算，而Bitmap引擎则负责生成数据的Bitmap索引，并将结果存储在存储当中。计算过程是基于预先定义好维度和度量的事实表来进行的。而某些不能预先进行索引的数据，则存储在列式数据库当中，从而可以在没有预先计算的情况下，高效的执行分析型的计算。 ## 大数据平台之进阶 随着公司业务的进一步发展和扩充，对于数据计算的需求已经不仅仅是统计分析类型的业务，对数据价值的探索变得越来越重要。面向统计分析业务的平台已经不能够满足业务发展的需求，我们需要对平台进行进一步重构，使得大数据平台能够满足： - 统计分析业务 - 交互式分析 - 机器学习 - 数据可视化 基于这些需求，TalkingData新的π系统应运而生。整个平台的架构如下图：  在新的π系统架构中，整个大数据平台除了能够支持统计分析业务，还增加了对机器学习、以及交互式分析的支持。不同的技术组件描述如下： ## Data Collector 在新的价格中，为了提高数据收集的效率，Data Collector在新的架构中从基于java servlet改为基于actor模型的node.js进行实现。另外Data Collector本身分为前置节点和中心节点两级，从而可以实现数据收集的分布式部署。前置节点分布式部署在多个区域，使得SDK可以选择网络连接更快的节点发送数据。而前置节点和中心节点采用高压缩比的数据传输，从而更好的利用中心机房的带宽资源。 ## Message Bus 在新的架构中，消息队列还是采用基于Kafka的消息总线，从而保证平行扩展、高可靠性，另外支持多消费者。 ## Storage Service 在新的架构中，我们将使用到的存储做为服务进行了封装。整个存储部分根据数据的冷热时间不同，进行分区。热数据存储在分布式缓存Tachyon中，而冷数据则以Parquet格式存储在HDFS当中。为了更好的支持多维交叉的分析型业务，TalkingData开发了针对bitmap的bitmap存储。所有的存储可以通过封装好的API进行统一的访问。另外，引入了基于HCatalog进行封装的元数据管理，从而方便对数据的管理和访问。 ## Compute Service 计算服务基于分布式计算框架Spark，其中融合了Torch系统中的bitmap引擎，从而可以对流式数据生成bitmap索引，并将索引存储在bitmap存储中。另外将流式消费的数据转化为列式存储结构，存储在Tachyon中。Tachyon中存储的数据有有效期，过期的数据会迁移到HDFS当中，并且在Tachyon中做清除。即时数据请求会根据请求类型和时间，决定是从bitmap存储、Tachyon、还是HDFS中读取数据。所有的数据计算封装为统一的数据计算API。 ## Machine Learning Service 为了更好的发挥数据的价值，我们的架构中引入了机器学习服务。机器学习服务包括了Spark提供的MLLib，另外也包括公司自己开发的一些高效的机器学习算法，比如随机决策森林、LR等等算法。所有的算法都封装为算法库，通过API的方式提供调用。 ## Query Engine 查询引擎则是对存储API、计算API和机器学习API进行封装，上层业务可以通过类似于SQL的语句进行数据计算，查询引擎会对查询进行解析，然后转化为对应的下层API调用和执行。 ## Data Service API 数据服务API则是各数据业务系统对数据进行业务化封装的API，这些API一般都是Restful API。数据可视化层可以通过这些Restful API获取数据，进行数据展现。 ## Data Visualization 数据可视化服务包含标准的数据可视化组件，通过对数据可视化组件化封装，业务系统的开发变得更为高效。数据可视化组件通过与数据服务API交互，获取需要的数据，完成数据的可视化展现。 ## 后记 新的π系统的架构是TalkingData技术团队第一次以更为面向全局的视角进行的一次架构重构。整个架构的设计和实现也融合了公司不同技术团队的集体力量，整个架构目前还在逐步完善中，期望我们能够将这个架构变得更加成熟，实现的更加灵活，变成一个真正的可平行扩展的支持多种大数据计算能力的大数据平台。