🌏【架构师指南】分布式技术知识点总结（数据处理）

数据分析

从传统的基于关系型数据库并行处理集群、用于内存计算近实时的，到目前的基于 hadoop 的海量数据的分析，数据的分析在大型电子商务网站中应用非常广泛，包括流量统计、推荐引擎、趋势分析、用户行为分析、数据挖掘分类器、分布式索引等等。

• 并行处理集群有商业的 EMC Greenplum，Greenplum 的架构采用了 MPP(大规模并行处理)，基于 postgresql 的大数据量存储的分布式数据库。

• 内存计算方面有 SAP 的 HANA，开源的 nosql 内存型的数据库 mongodb 也支持 mapreduce 进行数据的分析。

• 海量数据的离线分析目前互联网公司大量的使用 Hadoop、Spark、Blink、Flink。Hadoop 在可伸缩性、健壮性、计算性能和成本上具有无可替代的优势，事实上已成为当前互联网企业主流的大数据分析平台。

Hadoop 通过 MapReduce 的分布式处理框架，用于处理大规模的数据，伸缩性也非常好；但是 MapReduce 最大的不足是不能满足实时性的场景，主要用于离线的分析。

基于 MapRduce 模型编程做数据的分析，开发上效率不高，位于 hadoop 之上 Hive 的出现使得数据的分析可以类似编写 sql 的方式进行，sql 经过语法分析、生成执行计划后最终生成 MapReduce 任务进行执行，这样大大提高了开发的效率，做到以 ad-hoc(计算在 query 发生时)方式进行的分析。

1. 基于 MapReduce 模型的分布式数据的分析都是离线分析，执行上是暴力扫描，无法利用类似索引的机制；

2. 开源的 Cloudera Impala 是基于 MPP 的并行编程模型的，底层是 Hadoop 存储的高性能的实时分析平台，可以大大降低数据分析的延迟。

1. 目前 Hadoop 使用的版本是 Hadoop1.0，一方面原有的 MapReduce 框架存在 JobTracker 单点的问题，另外一方面 JobTracker 在做资源管理的同时又做任务的调度工作，随着数据量的增大和 Job 任务的增多，明显存在可扩展性、内存消耗、线程模型、可靠性和性能上的缺陷瓶颈；

2. Hadoop2.0 yarn 对整个框架进行了重构，分离了资源管理和任务调度，从架构设计上解决了这个问题。

实时计算

在互联网领域，实时计算被广泛实时监控分析、流控、风险控制等领域。电商平台系统或者应用对日常产生的大量日志和异常信息，需要经过实时过滤、分析，以判定是否需要预警；

同时需要对系统做自我保护机制，比如对模块做流量的控制，以防止非预期的对系统压力过大而引起的系统瘫痪，流量过大时，可以采取拒绝或者引流等机制；有些业务需要进行风险的控制，比如彩票中有些业务需要根据系统的实时销售情况进行限号与放号。

原始基于单节点的计算，随着系统信息量爆炸式产生以及计算的复杂度的增加，单个节点的计算已不能满足实时计算的要求，需要进行多节点的分布式的计算，分布式实时计算平台就出现了。

这里所说的实时计算，其实是流式计算，概念前身其实是 CEP 复杂事件处理，相关的开源产品如 Esper，业界分布式的流计算产品 Yahoo S4,Twitter storm、flink、blink 等，以 storm 和 blink 和 flink 开源产品使用最为广泛。

对于实时计算平台，从架构设计上需要考虑以下几个因素：

伸缩性

随着业务量的增加，计算量的增加，通过增加节点处理，就可以处理。

高性能、低延迟

从数据流入计算平台数据，到计算输出结果，需要性能高效且低延迟，保证消息得到快速的处理，做到实时计算。

可靠性

保证每个数据消息得到一次完整处理。

容错性

系统可以自动管理节点的宕机失效，对应用来说，是透明的。

整个集群的管理是通过 zookeeper 来进行的。客户端提交拓扑到 nimbus。

Nimbus 针对该拓扑建立本地的目录根据 topology 的配置计算 task，分配 task，在 zookeeper 上建立 assignments 节点存储 task 和 supervisor 机器节点中 woker 的对应关系。

zookeeper 上创建 taskbeats 节点来监控 task 的心跳；启动 topology。

Supervisor 去 zookeeper 上获取分配的 tasks，启动多个 woker 进行，每个 woker 生成 task，一个 task 一个线程；根据 topology 信息初始化建立 task 之间的连接;Task 和 Task 之间是通过 zeroMQ 管理的；之后整个拓扑运行起来。

Tuple 是流的基本处理单元，也就是一个消息，Tuple 在 task 中流转，Tuple 的发送和接收过程如下：

1. 发送 Tuple，Worker 提供了一个 transfer 的功能，用于当前 task 把 tuple 发到到其他的 task 中。以目的 taskid 和 tuple 参数，序列化 tuple 数据并放到 transfer queue 中。

2. 在 0.8 版本之前，这个 queue 是 LinkedBlockingQueue，0.8 之后是 DisruptorQueue。

3. 在 0.8 版本之后，每一个 worker 绑定 inbound transfer queue 和 outbound queue，inbound queue 用于接收 message，outbound queue 用于发送消息。

发送消息时，单个线程从 transferqueue 中拉取数据，把这个 tuple 通过 zeroMQ 发送到其他 worker 中。

接收 Tuple，每个 worker 都会监听 zeroMQ 的 tcp 端口来接收消息，消息放到 DisruptorQueue 中后，后从 queue 中获取 message(taskid,tuple)，根据目的 taskid,tuple 的值路由到 task 中执行。每个 tuple 可以 emit 到 direct steam 中，也可以发送到 regular stream 中，在 Reglular 方式下，由 Stream Group（stream id-->component id -->outbound tasks）功能完成当前 tuple 将要发送的 Tuple 的目的地。

通过以上分析可以看到，Storm 在伸缩性、容错性、高性能方面的从架构设计的角度得以支撑；同时在可靠性方面，Storm 的 ack 组件利用异或 xor 算法在不失性能的同时，保证每一个消息得到完整处理的同时。

实时推送

实时推送应用场景非常多，比如系统的监控动态的实时曲线绘制，手机消息的推送，web 实时聊天等。

实时推送有很多技术可以实现，有 Comet 方式，有 websocket 方式等。

Comet 基于服务器长连接的“服务器推”技术，包含两种：

• Long Polling：服务器端在接到请求后挂起，有更新时返回连接即断掉，然后客户端再发起新的连接

• Stream 方式:每次服务端数据传送不会关闭连接，连接只会在通信出现错误时，或是连接重建时关闭（一些防火墙常被设置为丢弃过长的连接， 服务器端可以设置一个超时时间， 超时后通知客户端重新建立连接，并关闭原来的连接）。

Websocket：长连接，全双工通信

是 HTML5 的一种新的协议。它实现了浏览器与服务器的双向通讯。webSocket API 中，浏览器和服务器端只需要通过一个握手的动作，便能形成浏览器与客户端之间的快速双向通道，使得数据可以快速的双向传播。

Socket.io 是一个 NodeJS websocket 库，包括客户端的 JS 和服务端的的 nodejs，用于快速构建实时的 web 应用。

数据存储

数据库存储大体分为以下几类，有关系型（事务型）的数据库，以 oracle、mysql 为代表，有 keyvalue 数据库，以 redis 和 memcached db 为代表，有文档型数据库如 mongodb，有列式分布式数据库以 HBase，cassandra,dynamo 为代表，还有其他的图形数据库、对象数据 库、xml 数据库等。每种类型的数据库应用的业务领域是不一样的，下面从内存型、关系型、分布式三个维度针对相关的产品做性能可用性等方面的考量分析。

内存型数据库

内存型的数据库，以高并发高性能为目标，在事务性方面没那么严格，以开源 nosql 数据库 mongodb、redis 为例。

Mongodb

通信方式

多线程方式，主线程监听新的连接，连接后，启动新的线程做数据的操作（IO 切换）。

数据结构

数据库-->collection-->record

• MongoDB 在数据存储上按命名空间来划分，一个 collection 是一个命名空间，一个索引也是一个命名空间。

• 同一个命名空间的数据被分成很多个 Extent，Extent 之间使用双向链表连接。

• 在每一个 Extent 中，保存了具体每一行的数据，这些数据也是通过双向链接连接的。

每一行数据存储空间不仅包括数据占用空间，还可能包含一部分附加空间，这使得在数据 update 变大后可以不移动位置。

索引以 BTree 结构实现。

如果你开启了 jorunaling 日志，那么还会有一些文件存储着你所有的操作记录。

持久化存储

• MMap 方式把文件地址映射到内存的地址空间，直接操作内存地址空间就可以操作文件，不用再调用 write,read 操作，性能比较高。

• mongodb 调用 mmap 把磁盘中的数据映射到内存中的，所以必须有一个机制时刻的刷数据到硬盘才能保证可靠性，多久刷一次是与 syncdelay 参数相关的。

• journal（进行恢复用）是 Mongodb 中的 redo log，而 Oplog 则是负责复制的 binlog。如果打开 journal，那么即使断电也只会丢失 100ms 的数据，这对大多数应用来说都可以容忍了。从 1.9.2+，mongodb 都会默认打开 journal 功能，以确保数据安全。而且 journal 的刷新时间是可以改变的，2-300ms 的范围,使用--journalCommitInterval 命令。Oplog 和数据刷新到磁盘的时间是 60s，对于复制来说，不用等到 oplog 刷新磁盘，在内存中就可以直接复制到 Sencondary 节点。

事务支持

Mongodb 只支持对单行记录的原子操作

HA 集群

用的比较多的是 Replica Sets，采用选举算法，自动进行 leader 选举，在保证可用性的同时，可以做到强一致性要求。

当然对于大量的数据，mongodb 也提供了数据的切分架构 Sharding.

部署平台

监控、统计

大型分布式系统涉及各种设备，比如网络交换机，普通 PC 机，各种型号的网卡，硬盘，内存等等，还有应用业务层次的监控，数量非常多的时候，出现错误的概率也会变大，并且有些监控的时效性要求比较高，有些达到秒级别；在大量的数据流中需要过滤异常的数据，有时候也对数据会进行上下文相关的复杂计算，进而决定是否需要告警。因此监控平台的性能、吞吐量、已经可用性就比较重要，需要规划统一的一体化的监控平台对系统进行各个层次的监控。

平台的数据分类

• 应用业务级别：应用事件、业务日志、审计日志、请求日志、异常、请求业务 metrics、性能度量

• 系统级别：CPU、内存、网络、IO

• 时效性要求

• 阀值，告警：

• 实时计算：

• 近实时分钟计算

• 按小时、天的离线分析

• 实时查询

节点中 Agent 代理可以接收日志、应用的事件以及通过探针的方式采集数据，agent 采集数据的一个原则是和业务应用的流程是异步隔离的，不影响交易流程。

• 数据统一通过 collector 集群进行收集，按照数据的不同类型分发到不同的计算集群进行处理；

• 有些数据时效性不是那么高，比如按小时进行统计，放入 hadoop 集群；

• 有些数据是请求流转的跟踪数据，需要可以查询的，那么就可以放入 solr 集群\ES 集群进行索引；

• 有些数据需要进行实时计算的进而告警的，需要放到 storm 集群中进行处理。

数据经过计算集群处理后，结果存储到 Mysql 或者 HBase 中。

监控的 web 应用可以把监控的实时结果推送到浏览器中，也可以提供 API 供结果的展现和搜索。