七牛大数据平台架构演进之路与使用实战
看着图1大家可能会感到熟悉，又或者会觉得部分有些陌生，这是一张汇集了目前大数据生态下大多数成熟组件的架构图。众所周知，大数据生态很复杂，对于个人来说，要全部学会可能要花费好几年时间。而对于企业来说，要最大程度发挥其价值，构建一个成熟稳定、功能多样的大数据平台，期间花费的时间以及人力成本着实难以估量，更何况还需要考虑持续维护的问题。这就是七牛的 Pandora 大数据平台灵感的来源，我们构建一个大数据平台，作为产品提供给用户，快速帮助用户挖掘数据价值。
七牛是以云存储起家的公司，平台上有着大量的数据、业务日志以及运维监控数据，基于对这些数据的管理以及分析的需求，Pandora 诞生了。我们搭建了一个可靠的大数据平台，将大数据生态中的各个组件配套成一个体系发挥作用，用来解决实际业务中碰到的繁琐、复杂、多样化的问题。这个大数据平台的工作从数据的采集便已开始，通过一条数据总线，再根据业务需求分流到不同的下游产品，从存储到数据可视化，从实时的数据变换到离线的算法分析，构建一个全栈的大数据分析产品。
与此同时，我们在大数据平台之上构建了业务工作流的概念，让用户只需关心构建符合他们业务模型的工作流，而无需具备大数据技术背景。不仅大大降低了用户挖掘大数据价值的成本，更为重要的是去除了大数据技术门槛，使得各行各业的专家可以更好的施展自己对业务的深度理解。
在工作流上，用户不仅可以清晰的监控自己的数据流动，轻松构建各类实时、离线的数据变化与自定义计算，还可以按需弹性扩容、快速调度云端资源，降低了运维的成本。与此同时，我们集成了社区中大量的优秀开源软件，对其进行优化及定制，一方面以便发挥其更强大的功能，另一方面，也让熟悉使用这类开源软件的用户可以做到快速迁移，无缝切换使用。
那么，Pandora 到底是一个怎样的平台呢？工作流又是怎样的呢？首先来直观的看一下工作流的使用形态，如图 2 所示。
最左边的数据源是工作流的起点，数据源可以是一个，也可以是多个。在实时计算的工作流中，我们只能有一个数据源，这个数据源就是数据收集汇聚的中心，也可以理解为数据总线，所有不同终端的数据打向数据源，再通过数据源根据业务需求分发到不同下游；在离线工作流中，我们可以有多个数据源，不同的数据源代表的是存储在不同地方的离线数据，可以是七牛云存储中的不同文件，又或是 HDFS 等不同类型的数据仓库。
不管是实时还是离线，从数据源开始，你就可以根据需要做不同类型的处理。
最基本的处理是对数据进行一些定制化的计算，比如你可能需要对每天海量的数据进行一个定时分析汇聚的功能，计算每分钟有多少条数据、每小时有多少条数据，从而缩减数据规模节约存储成本，或者从中生成一份数据日报、周报等等；又比如在这个信息爆炸的时代，你从网上抓取了海量的数据，需要对数据进行一些清洗、过滤、删选，以此分析社会热点或其他有价值的信息；又比如你想对数据做一个延伸或扩展，最常见的就是对一个 IP 获取它的运营商、所在区域等信息。那么你就可以创建一个计算任务，最简单的编写一些 SQL 语句就可以做数据变换；进阶一些的使用方式就是编写一些 UDF(用户自定义的函数)，做一些较为复杂的变化；更高阶的就是直接编写一类插件，直接使用大量 Java 的类库对数据进行操作。当然，在离线计算中，除了单个数据源的计算任务以外，你还可以对两个数据源，亦或是两个计算任务的结果进行聚合，然后再进行计算等等。计算任务可以满足你对于整个工作流的完整数据处理需求。
在进行过一个基本的计算以后，可能最常见的一个需求就是对这些计算后的数据进行检索，直白的说就可以查询你的数据。那么你可以创建一个导出到日志检索，在这里你就可以搜索你的计算结果。当然，你的数据在数据源中也完全可以不经过任何计算任务，直接导向日志检索。又或者你希望对数据进行更完善的实时监控和数据可视化处理，那么就可以导出到时序数据库，针对带有时间戳的数据做高性能数据写入和查询优化，满足你针对实时海量数据的即席查询需求。
另一方面，你工作流计算后的结果，可以直接再次导出到七牛云存储进行永久保存，或者与后续的数据结合进行分析。你也可以理解为通过大数据服务，七牛的云存储变成了一个数据仓库为客户提供服务。之前已经存储在七牛云上的数据 (如 CDN 日志等)，完全可以直接使用这个大数据平台进行计算，无需任何接入过程。
为了方便用户充分利用自己的数据，我们提供了导出到 HTTP 服务，用户可以构建自己的 HTTP 服务器来接受经过 Pandora 大数据平台计算后的数据。
图 3 是 Pandora 的产品架构图，基本的功能在第 2 节中均已介绍，在此不再赘述，在讲解系统架构之前，让我们对照产品架构进行一些基本的系统组件技术名称的对照说明，以便下文描述简洁便于理解。数据通过数据上报工具 logkit、各类 SDK 或者用户直接调用开放 API 接入，数据进入后无论是数据源本身还是经过计算任务后的临时数据存储节点，我们都一称作消息队列，技术上称之为 Pipeline，像不同下游提供导出服务的组件我们称之为 Export，在 Pipeline 中承担各类计算任务处理的组件我们称之为 Transform，下游的时序数据库服务我们称之为 TSDB，下游的日志检索服务我们称之为 LogDB。
有了这些基本概念后，让我们对照图 4 Panora 系统架构图，开启Pandora 架构演进之旅。
最左侧的组件是数据收集的部分，数据来源于客户各种各样的系统。相信大部分用户在接入大数据平台时，都会面临数据收集这一难题，一方面希望数据不重不漏全部收集上来，一方面又要数据收集尽可能高效且不太消耗机器的各类资源，同时还要满足场景各异的不同情况下的数据收集需求。熟悉这块的朋友可能也早已了解，社区已经有很多不同类型的开源数据收集工具，知名的包括 flume、logstash、fluentd、telegraf 等等，他们各有利弊，功能上大都满足需求，但是在性能上和一些非通用需求的场景上不尽如人意。为了更好的满足用户不同类型的需求，我们自主研发了一个可以收集各种各样数据源数据的工具 logkit，图 5 是 logkit 的功能架构示意图。logkit 使用 go 语言编写，以插件的形式定制不同的数据收集解析需求，处理高效、性能损耗低，同时也已经开源，非常欢迎大家一起参与到 logkit 的使用和代码开发定制中来，为 logkit 提提 PR，当然，也非常乐意接受关于 logkit 的任何意见或建议，只需在 github 提 issues 即可。
有了这样一款数据收集工具，几乎 90% 的数据收集场景已经解决了，但是还会有诸如 ios、android 客户端数据直接上报、页面请求点击数据直接上报等需求，为此我们提供了各类语言的 SDK 方便用户使用，以弥补 logkit 功能上无法满足的需求。
数据收集上来后，就正式进入Pandora 大数据平台了。所有上报的数据无论最终是否计算或存储，都会统一暂存进入大数据总线 Pipeline。相信经过上面的介绍，很多读者早已发现，Pandora 帮助用户根据不同场景选择最适合的大数据分析方式。而这套模式的核心，毋庸置疑，就是处理数据在不同大数据产品间的流转。
Pipeline 就是这样一条数据总线，在数据总线的基础上我们打通一条条管，根据所需的场景导出到后端相应的存储服务上。同时据此来进行资源分配和任务管理。这样一来，就可以避免用户技术选型及技术架构与使用姿势和业务场景不匹配的情况，同时也可以利用云计算的优势，按需分配、快速扩容。
如图 6 所示是我们的第一版架构，实现上通过定制开源版本的 confluent，并把它作为这个架构系统的核心。数据要流入系统，我们首先构建了一个 Points Gate（API 服务器），Points Gate 解析校验用户的数据格式并调用 confluent 中 kafka-Rest 提供的 rest API 将数据写入到 kafka，利用 schema-registry 完成数据格式的校验以及数据解析，通过 kafka 获得数据管道的能力。
在进行元数据创建时，调度器在元数据服务器上创建一个用户元数据存储在 MongoDB 当中。对于 MongoDB 的元数据访问，我们构建了一个二级缓存系统（即图中 qconf），数据在进入或者导出时都会通过二级缓存访问元数据，这样数据就可以快速得到校验，扛住海量的数据吞吐。Kafka 本身包含了 Zookeeper 组件，我们也借此来保证整体系统组件的服务发现以及数据一致性这两个问题。
然而，随着应用的增加，数据量越来越大，这样，单个定制版的 Confluent 并不能满足这些数据量增长的业务压力，以及用户不断增加的场景需求。kafka topic(partition) 不断增长导致整体响应变慢，无法快速切换灾备等待问题日益凸显。在这个基础上，我们对原本的系统架构进行了调整。
如图 7 所示，我们对 Pipeline 的第一次整体升级包含了大量的组件基础架构的调整。首先我们构建了 Confluent 的多集群系统，将单个 Confluent 集群规模控制在 100 台机器以内，分区数量控制在 1 万以内，根据需求对集群进行路由。
可见通过多集群系统，我们可以合理控制每个 confluent 集群的规模，同时根据我们的调度器按照需要切换用户到不同的集群上，实现快速切换、扩容、容灾、隔离等调度需求。
其次我们对 Points Gate、Transform、Export 中无状态组件进行了容器化处理，利用七牛内部的容器云平台，实现了无状态服务的快速部署、扩容以及灰度发布等需求。
这次架构的调整效果显著，我们成功抗住了每天上百 TB，千亿级数据点的数据增量。
不止于此，为了更高的性能提升以及节约成本，我们在上述升级之后进行了第二次的架构升级。这次升级主要体现在对 Confluent 的进一步定制（或者说替换），不再使用 kafka-rest，同时对打点的数据格式进一步优化，又一次节约了近一半的机器成本。
在解决了数据总线问题以后，问题的重中之重自然是如何处理数据导出的问题。众所周知，数据导出其实就是从一个上游系统拉取数据，然后将数据再发送到下游系统的过程。但这里面涉及的难点和调整可能大多数都是不为人知的了。在介绍导出服务架构之前，非常有必要描述一下针对海量数据导出，到底有哪些挑战？
首先面对的第一大挑战自然是 高吞吐量 的问题，海量数据不断涌入带来的必然问题就是网卡和 CPU 分配的问题，一旦流量分配不均，就会出现大量因网卡、CPU 负载过高带来的延迟，严重影响服务可用性。
显然，保证低延迟 就是第二个挑战，一旦各个链路有一个环节配合不均衡，就会带来大量延迟，如何保证导出的上下游始终保持较高的吞吐量，从而保证较低的延迟，是一个非常大的调整。
为了保证低延迟，就要更好地 适配多种下游，使其始终保证高吞吐，了解下游不同服务的特性，并针对这些特性动态的调整资源，亦是一个很大的挑战。
除此之外还有分布式系统的常见问题，需要保证 服务高可用，以及 水平扩展 的能力。保证任务单元标准化，任务粒度可以切分扩展；保证调度任务节点故障不影响其他节点的正常导出等等。
最为重要的是 自动化运维，当导出的任务涵盖数十上百台机器后，人力已经无法精细化处理每台机器的任务分配，资源必须可以自动调度、调整以及构建统一的任务监控。
让我们来看一下导出服务的功能架构图，如图 8 所示。我们的导出服务主要涉及三个层级，一个是元数据管理，在这一层保证任务的分配以及监控展示；第二层则是任务管理层，除了基本的任务切分、并发管理以及通信协议之外，还包含了压力预估模块，根据之前的数据量预估下一阶段的数据流量，从而调整资源分配；再下一层则是数据处理层，在这一层完成诸如数据预取、数据校验、压缩以及推送到下游等任务。
在最初的版本中，我们会在 zookeeper 上面创建一个任务 (task)，Export 通过分布式锁对 task 进行争抢，被抢到的任务则开始直接导出，如图 9 所示。
在这样一个初步架构中，我们基本完成了水平扩展以及高可用的需求，同时做了诸如数据预取，延迟报警、数据监控等多种功能和优化。但是流量上来以后，很容易出现某个机器争取的任务流量变大，导致大量数据打到同一台机器上，导致网卡和 CPU 负载过高，延迟急剧升高。本质上就是流量分布不均匀，导致导出性能低下，机器资源的平均利用率也低。
此时，我们对该方案进行第一次架构升级，如图 10 所示。我们将原来 topic 级别的任务按照 parition 进行分布式消费。为了使得每个 partition 粒度的任务大体是均等的，我们将 partition 承载的数据量按照标准化处理，并根据历史流量进行预测，预测结果超过当前我们定制的标准符合的对应容量即触发扩容，这样的标准化有效简化了调度的难度。
同时我们将原来纯粹的 export 改为 master/worker 结构，Master 对收集到的任务进行主动权衡分配，根据任务的历史流量进行流量预测、对任务的 partition 数量、每个 export worker 的机器资源剩余情况，进行综合调度。对于一些特殊任务做机器黑白名单绑定等功能。
在做了上述工作以后，机器的整体利用率有了很大的提升，但是由于下游系统的不同，写入吞吐量始终参差不齐，无法始终保持在一个较高的水平。为了解决该问题，我们再次对架构进行小范围升级，如图 11 所示，我们在导出的 export worker 端增加了一套对下游系统的适配加速模块。其核心思路就是按照下游的吞吐能力进行自动调节请求体大小以及并发度。这个主要是为了解决上下游传输数据速度不匹配，以及下游吞吐量不稳定的问题。
类似于 Flume 的思想，我们构建了一个内存队列，以事务的形式从队列中获取数据（或者失败回滚），根据下游的情况调整单次数据请求的大小和并发度，以及调整出错等待时间等。这样一来，整个导出的吞吐量就可以很有效的进行控制，去除了毛刺，极大的提高了机器资源的使用率以及导出效率。
解决了数据的导出问题，基本上绝大部分数据流转的问题也都解决了。下面开始关注 Pandora 下游的一系列服务。
TSDB 是七牛自主研发的分布式时序数据库服务。TSDB 针对时序数据定制存储引擎，根据时序数据带有时间戳的特性，我们针对时间戳做特殊的索引，实现数据高速汇入和实时查询获取的能力；同时构建了简单且高性能的 HTTP 写点和查询接口，为查询聚合数据量身定制了类 SQL 语言，完全兼容开源社区 InfluxDB 的 API，支持无缝对接到 Grafana，对数据进行高比例压缩，实现低成本存储。除此之外，TSDB 拥有开源社区版本的 InfluxDB 所没有的分布式、多集群、高可用，水平扩容、以及分库分表能力。
如图 12 所示，TSDB 是我们基于 tsm 构建的分布式时序数据库，拥有每秒 60 万条记录的写入性能以及实时查询聚合分析的能力。在分布式方面，除了基本的多集群、多租户隔离的概念以外，我们还针对性的做了两个强大的扩容功能，一个是根据时序进行纵向集群切割，解决海量数据写入时磁盘的快速扩容问题；另一个则是根据用户的标签进行数据库表横向切割，类似传统数据的分库分表能力。在进行了这两大扩展能力变换后，数据的写入和查询依旧高效，甚至查询速度在分库分表后性能有所提升。
为了实现这样的扩容功能，我们基于此构建了一个分布式计算引擎，解析用户的 SQL 并变成一个个执行计划，将执行计划下推至不同的 TSM 计算引擎实例中进行计算，最后再进行数据 reduce 计算反馈给用户。
除了数据写入性能高以外，还支持数据即时查询，数据写入成功即可查询，数据零延迟；同时支持 InfluxDB 的持续聚合功能，类似于定时任务一样将持续写入的数据不断进行聚合计算；当单个用户数据量过大时，拥有横向拓展能力，集群扩展后写入性能不打折，查询效率更高。针对时序数据的特性，我们将数据进行冷热分离，对数据按照时间分片，使最近的数据查询性能更高。
在了解完时序数据库以后，让我们来看一下下游的另一大服务，日志检索服务，又称 LogDB。日志搜索其实是几乎所有技术开发人员都会需要的服务，传统解决方案 (ELK，Elasticsearch、Logstash、Kibana) 针对小数据量不会出现任何问题。但是当数据量过于庞大时，这些方案也就不那么适用了。
我们 LogDB 的底层可以通过插件的形式接入不同类型的搜索引擎，包括 Solr、Elasticsearch（简称 ES）等，目前承载海量数据搜索任务的底层引擎主要使用的是 ES。与单纯的使用 ES 不同，LogDB 本身是一套分布式系统，管理的单元可以是一个 ES 节点，也可以是一个 ES 集群，所以我们构建了大量的 ES 集群，不同的集群用以适配不同的用户以及不同的搜索需求。
大体上我们将搜索的需求分为两类，一类是 ELK 类需求，针对如程序运行日志、业务访问日志等收集索引，这类需求的普遍特点是数据量大，时效性高，带有时间戳，无需存储太长时间，无需更新；另一类需求类似于搜索引擎，数据存在更新需要，且强依赖于不同类型的分词器，数据冷热不明显，不带有明显的时间属性，我们称之为通用检索需求。这两类需求，LogDB 都是完全支持的，但是针对这两类需求，我们做的优化不同。
在我们讨论具体的优化之前，先来看一下 LogDB 的架构图， 如图 13 所示。首先是数据的写入，LogDB 是 Pandora 平台下游服务，上游主要是之前提到的 Pipeline 以及 Export。Export 导出的数据通过 apisever 将数据导入到不同的 ES 集群当中，根据不同用户的需求给他们提供不同的集群服务，集群之间也可以相互进行切换。
那么如何确认数据到底数据哪个集群呢？为了使得海量的数据快速确认，我们对元数据进行了多级缓存，包括 MongoDB 的实际存储、memcached 的二级缓存，以及本地的缓存，极大提高了数据校验的速度。除此之外，LogDB 本身也是 Pandora 的用户，使用了 TSDB 对自身数据进行监控，使用七牛云存储进行数据快照与备份。同时将计费数据导出到云存储，利用 XSpark 机器进行离线计算。
架构中剩下的部分都是我们针对不同索引做的优化。简而言之，我们首先构建了一个高性能队列，可以让外部的数据持续高吞吐写入；同时我们还会根据历史流量进行动态索引平衡、不同集群的索引跨集群平衡、索引定时清理、VIP 集群隔离等功能；并且会对 ES 的搜索进行分步搜索控制，缓存历史搜索，优化用户搜索的效率和体验等等。关于这一块的详细内容，我们团队在之前已经分享过，可以阅读这篇 《基于 Elasticsearch 构建千亿流量日志搜索平台实战》公众号文章了解更详细的内容。
XSpark 不仅与 Pandora 整体完全打通，可以把七牛云存储当成一个数据仓库使用，又完全可以独立使用。即用户除了在 Pipeline 里面做离线计算之外，你还可以选择直接一键生成一个基于容器云的个人专属 Spark 集群，直接读取你自己的数据源，只要 Spark 支持的数据格式，XSpark 都支持。如果你的数据已经存储在七牛云存储上，XSpark 可以直接高效读取并计算，XSpark 是 Pandora 提供给大数据高端用户的一个高度灵活的离线计算产品。
显然，容器云所具有的优势 XSpark 全都具备，你完全可以根据需要动态伸缩的 XSpark 资源数量与规格，按需启停等等。
图 14 是 XSpark 的架构图。我们将 Spark 的 master 和 worker 分为不同的容器，首先启动 Spark 的 master 容器，获取 master 地址，然后再根据用户的配置，启动相应数量的 worker 容器，worker 容器自动向 master 注册。同时容器云的支撑使得XSpark 可以在运行过程中进行缩容扩容。
同时 XSpark 也开放了完整的 Spark 监控以及管理控制页面，完全兼容开源社区的 Zepplin 使用方式。
那么这样的一套系统，用户体验到底是怎么样的呢？下面以收集最常见的 nginx 日志为例来看下实际体验。
首先，为了让用户最舒适的将数据上传到七牛平台，用户可以直接部署我们的 logkit 工具，然后打开 logkit 配置页面，如图 15 所示。
根据需要一步步选择数据源、配置解析方式、发送方式最后确认并添加。当然，在配置解析方式的页面，你还可以通过页面预先尝试解析器配置是否正确，如图 16 所示。
通过简单的几步配置后，你的数据便已经上传到 Pandora 平台，并自动生成了工作流，并且创建好了导出到日志检索服务，如图 17 所示。
在工作流中，你还可以创建一些计算任务做一些分析计算，导出到云存储作为数据计算，导出到时序数据库进行复杂的 SQL 查询。之后的大数据分析工作也就会相对容易一些。有了这些准备工作，我们就可以开始进行日志检索查询（图 18 所示）、数据可视化（如图 19 所示）、离线分析（如图 20 所示）等工作。
在我们这仅仅两三步简单的操作以后，就可以看到哪些十分有价值的数据呢？
实时总用户访问量 (请求数统计)，如图 21 所示
机器请求数随时间变化趋势，如图 22 所示
实时请求状态码占比，如图 23 所示
实时请求 TOP 排名，如图 24 所示
实时请求来源 IP TOP 排名，如图 25 所示
响应时间随时间变化趋势图，如图 26 所示
实时用户请求的客户端 TOP 排名，如图 27 所示
实时根据不同情况进行具体数据的查询，包括状态码、响应时间范围进行筛选等，如图 28 所示
其他更多自定义配置...
自定义的 Grafana DashBoard 配置示例，如图 29 所示
可见，仅仅几步简单的操作，你就借助 Pandora 实现了海量日志的实时监控，通过 nginx 日志完整而详尽地了解业务的流量入口的各类情况。
有了实时的数据监控，怎么能少得了报警呢，我们还提供了包括 Slack、Email 邮箱、Webhook 等十来种报警方式。
比如说如图 30，我们设置了一个响应时间大于 1000ms 的报警
LogDB 采用的是基于 Elasticsearch 协议的报警，这个基于 Elasticsearch 协议的 Grafana 报警功能是七牛独家。TSDB 是基于 InfluDB 协议，使用 TSDB 同样具备上述功能。
那么您可以看到的报警形式是怎么样的呢？
看到图 31 Slack 上的报警了吗？除了基本的文字，还会带上酷炫的报警图片，图片都会被存储到您七牛云存储的 bucket 里面。
如图 32 所示，常规的邮件报警内容也一样酷炫！
如果你使用离线分析，那么可以获得的内容就更多了，只要是数据拥有的维度，就可以统计出来，比如以 CDN 日志为例。
用户地区分布，如图 33 所示
用户机型分布，如图 34 所示
活跃用户数随时间变化趋势，如图 35 所示
各类手机图片下载数，如图 36 所示
不同时间平均下载速度，如图 37 所示
不同时间正常响应比重，如图 38 所示
查询用户系统分布，如图 39 所示
除了上述这些，还支持更细粒度的下钻功能，可以全方位无死角分析你的海量数据。
至此，本次的 Pandora 大数据平台的架构演进与使用实战算是介绍完了，但是Pandora 大数据产品的迭代还在不断向前，还有大量大数据平台建设的细节等待我们去完善。
今日荐文
点击下方图片即可阅读
有赞客户行为收集与实时处理系统设计解析
相关内容
微信扫码咨询专知VIP会员