大数据技术原理与应用 - (11). 流计算

【第三篇】 - 大数据处理与分析, 《大数据技术原理与应用, 林子雨》

本篇介绍大数据处理与分析的相关技术，包括

• 第7章 - MapReduce
• 第8章 - Hive - 基于 Hadoop 的数据仓库
• 第9章 - Hadoop 的优化与发展
• 第10章 - Spark
• 第11章 - 流计算
• 第12章 - 图计算

大数据包括批量计算和流计算，不同于批数据处理，流式计算 (处理) 要求对数据流进行计算，要求更低的时延或实时结果输出。

流计算概述

静态数据和流数据

静态数据

如下图所示，很多企业为了支持决策分析而构建的数据仓库系统，其中存放的大量历史数据就是静态数据。技术人员可以利用数据挖掘和OLAP (On-Line Analytical Processing) 分析工具从静态数据中找到对企业有价值的信息。

流数据

近年来，在 Web 应用、网络监控、传感监测等领域，兴起了一种新的数据密集型应用——流数据，即数据以大量、快速、时变的流形式持续到达。

• 实例：PM2.5检测、电子商务网站用户点击流
• 特点：
  • 数据快速持续到达，潜在大小也许是无穷无尽的。
  • 数据来源众多，格式复杂。
  • 数据量大，但是不十分关注存储，一旦经过处理，要么被丢弃，要么被归档存储。
  • 注重数据的整体价值，不过分关注个别数据。
  • 数据顺序颠倒，或者不完整，系统无法控制将要处理的新到达的数据元素的顺序。

批量计算和实时计算

• 批量计算以“静态数据为对象”，可以在充裕的时间内对海量数据进行批量处理，如 Hadoop。
• 流数据不适合采用批量计算，因为流数据不适合用传统的关系模型建模。因为无法把远远不断的流数据保存到数据库中，流数据被处理后，一部分进入数据库成为静态数据，一部分直接丢弃。
• 流数据必须采用实时计算，响应时间为秒级。

流计算概念

流计算：实时获取来自不同数据源的海量数据，经过实时分析处理，获得有价值的信息。

流计算秉承一个基本理念，即数据的价值随着时间的流逝而降低，如用户点击流。因此，当事件出现时就应该立即进行处理，而不是缓存起来进行批量处理。为了及时处理流数据，就需要一个低延迟、可扩展、高可靠的处理引擎。

对于一个流计算系统来说，它应达到如下需求：高性能；海量式；实时性；分布式；易用性；可靠性。

流计算框架

• 商业级：IBM InfoSphere Streams 和 IBM StreamBase
• 开源流计算框架:
  • Twitter Storm: 免费、开源的分布式实时计算系统，可简单、高效、可靠地处理大量的流数据
  • Yahoo! S4 (Simple Scalable Streaming System): 开源流计算平台，是通用的、分布式的、可扩展的、分区容错的、可插拔的流式系统
• 公司为支持自身业务开发的流计算框架:
  • Facebook Puma
  • Dstream (百度)
  • 银河流数据处理平台 (淘宝)

流计算处理流程

• 传统的数据处理流程，需要先采集数据并存储在关系数据库等数据管理系统中，之后由用户通过查询操作和数据管理系统进行交互。
  传统的数据处理流程隐含了两个前提：

  • 存储的数据是旧的。存储的静态数据是过去某一时刻的快照，这些数据在查询时可能已不具备时效性了。
  • 需要用户主动发出查询来获取结果。

• 流计算的处理流程一般包含三个阶段：数据实时采集、数据实时计算、实时查询服务。

1. 数据实时采集

   • 数据实时采集阶段通常采集多个数据源的海量数据，需要保证实时性、低延迟与稳定可靠。
   • 以日志数据为例，由于分布式集群的广泛应用，数据分散存储在不同的机器上，因此需要实时汇总来自不同机器上的日志数据。

   • 目前有许多互联网公司发布的开源分布式日志采集系统均可满足每秒数百MB的数据采集和传输需求，如：

     • Facebook 的 Scribe
     • LinkedIn 的 Kafka
     • 淘宝 的 Time Tunnel
     • 基于 Hadoop 的 Chukwa 和 Flume

   • 数据采集系统的基本架构一般有以下三个部分：

     • Agent：主动采集数据，并把数据推送到 Collector 部分。
     • Collector：接收多个 Agent 的数据，并实现有序、可靠、高性能的转发。
     • Store：存储 Collector 转发过来的数据（对于流计算不存储数据）。

2. 数据实时计算
   • 数据实时计算阶段对采集的数据进行实时的分析和计算，并反馈实时结果。
   • 经流处理系统处理后的数据，可视情况进行存储，以便之后再进行分析计算。在时效性要求较高的场景中，处理之后的数据也可以直接丢弃。

3. 实时查询服务
   • 实时查询服务：经由流计算框架得出的结果可供用户进行实时查询、展示或储存。
   • 传统的数据处理流程，用户需要主动发出查询才能获得想要的结果。而在流处理流程中，实时查询服务可以不断更新结果，并将用户所需的结果实时推送给用户。
   • 虽然通过对传统的数据处理系统进行定时查询，也可以实现不断地更新结果和结果推送，但通过这样的方式获取的结果，仍然是根据过去某一时刻的数据得到的结果，与实时结果有着本质的区别。

可见，流处理系统与传统的数据处理系统有如下不同：

• 流处理系统处理的是实时的数据，而传统的数据处理系统处理的是预先存储好的静态数据。
• 用户通过流处理系统获取的是实时结果，而通过传统的数据处理系统，获取的是过去某一时刻的结果。
• 流处理系统无需用户主动发出查询，实时查询服务可以主动将实时结果推送给用户。

开源流计算框架 Storm

Twitter Storm 是一个免费、开源的分布式实时计算系统，Storm 对于实时计算的意义类似于 Hadoop 对于批处理的意义，Storm 可以简单、高效、可靠地处理流数据，并支持多种编程语言。Storm 框架可以方便地与数据库系统进行整合，从而开发出强大的实时计算系统。

Storm 的特点

Storm 可用于许多领域中，如实时分析、在线机器学习、持续计算、远程 RPC、数据提取加载转换等。

Storm 具有以下主要特点：

• 整合性: Storm 可方便地与队列系统和数据库系统进行整合。
• 简易的 API: Storm 的 API 在使用上既简单又方便。
• 可扩展性: Storm 的并行特性使其可以运行在分布式集群中。
• 容错性: Storm 可以自动进行故障节点的重启，以及节点故障时任务的重新分配。
• 可靠的消息处理: Storm 保证每个消息都能完整处理。
• 支持各种编程语言: Storm 支持使用各种编程语言来定义任务。
• 快速部署。
• 免费、开源。

Storm 设计思想

Storm 对一些设计思想进行了抽象化，其主要术语包括 Streams、Spouts、Bolts、Topology 和 Stream Groupings。

1. Streams: Storm 将流数据 Stream 描述成一个无限的 Tuple 序列，这些 Tuple 序列会以分布式的方式并行地创建和处理。
   • 每个 tuple 是一堆值，每个值有一个名字，并且每个值可以是任何类型。
   • Tuple 本来应该是一个 Key-Value 的 Map，由于各个组件间传递的 tuple 的字段名称已经事先定义好了，所以 Tuple 只需要按序填入各个 Value，所以就是一个Value List（值列表）。

2. Spout: Storm 认为每个 Stream 都有一个源头，并把这个源头抽象为 Spout。

   • 通常 Spout 会从外部数据源（队列、数据库等）读取数据，然后封装成 Tuple 形式，发送到 Stream 中。Spout 是一个主动的角色，在接口内部有个 nextTuple 函数，Storm 框架会不停的调用该函数。

3. Bolt: Storm 将 Streams 的状态转换过程抽象为 Bolt。Bolt 即可以处理 Tuple，也可以将处理后的 Tuple 作为新的 Streams 发送给其他 Bolt。

   • Bolt 可执行过滤、聚合、查询等操作。
   • Bolt 是一个被动的角色，其接口中有一个 execute(Tuple input) 方法，在接收到消息之后会调用此函数，用户可以在此方法中执行自己的处理逻辑。

4. Topology: Storm 将 Spouts 和 Bolts 组成的网络抽象成 Topology，它可以被提交到 Storm 集群执行。Topology 可视为流转换图，图中节点是一个 Spout 或 Bolt，边则表示 Bolt 订阅了哪个 Stream。当 Spout 或者 Bolt 发送元组时，它会把元组发送到每个订阅了该 Stream 的 Bolt 上进行处理。
   • Topology 里面的每个处理组件（Spout 或 Bolt）都包含处理逻辑， 而组件之
     间的连接则表示数据流动的方向
   • Topology 里面的每一个组件都是并行运行的。
   • 在 Topology 里面可以指定每个组件的并行度，Storm 会在集群里面分配那么多的线程来同时计算。
   • 在 Topology 的具体实现上，Storm 中的 Topology 定义仅仅是一些 Thrift 结构体（二进制高性能的通信中间件），支持各种编程语言进行定义。

5. Stream Groupings: Storm 中的 Stream Groupings 用于告知 Topology 如何在两个组件间（如 Spout 和 Bolt之间，或者不同的 Bolt 之间）进行 Tuple 的传送。每一个 Spout 和 Bolt 都可以有多个分布式任务，一个任务在什么时候、以什么方式发送 Tuple 就是由 Stream Groupings 来决定的。
   • 目前，Storm 中的 Stream Groupings 有如下几种方式：
     (1). ShuffleGrouping: 随机分组，随机分发 Stream 中的 Tuple，保证每个 Bolt 的 Task 接收 Tuple 数量大致一致。
     (2). FieldsGrouping: 按照字段分组，保证相同字段的 Tuple 分配到同一个 Task 中。
     (3). AllGrouping: 广播发送，每一个 Task 都会收到所有的 Tuple。
     (4). GlobalGrouping: 全局分组，所有的 Tuple 都发送到同一个 Task 中。
     (5). NonGrouping: 不分组，和 ShuffleGrouping 类似，当前 Task 的执行会和它的被订阅者在同一个线程中执行。
     (6). DirectGrouping: 直接分组，直接指定由某个 Task 来执行 Tuple 的处理。

Storm 框架设计

Storm 运行任务的方式与 Hadoop 类似：Hadoop 运行的是 MapReduce 作业，而 Storm 运行的是 “Topology”，但两者的任务大不相同，主要的不同是：MapReduce 作业最终会完成计算并结束运行，而 Topology 将持续处理消息（直到人为终止）。

Storm 集群采用 “Master—Worker” 的节点方式：

• Master 节点运行名为 “Nimbus” 的后台程序（类似 Hadoop 中的 “JobTracker”），负责在集群范围内分发代码、为 Worker 分配任务和监测故障
• Worker 节点运行名为 “Supervisor” 的后台程序，负责监听分配给它所在机器的工作，即根据 Nimbus 分配的任务来决定启动或停止 Worker 进程，一个 Worker 节点上同时运行若干个 Worker 进程

Storm 使用 Zookeeper 来作为分布式协调组件，负责 Nimbus 和多个 Supervisor 之间的所有协调工作。借助于 Zookeeper，若 Nimbus 进程或 Supervisor 进程意外终止，重启时也能读取、恢复之前的状态并继续工作，使得 Storm 极其稳定。

Worker 进程

1. Worker 进程: 每个 worker 进程都属于一个特定的 Topology，每个 Supervisor 节点的 worker 可以有多个，每个 worker 对 Topology 中的每个组件（Spout 或 Bolt）运行一个或者多个 executor 线程来提供task的运行服务。
2. Executor：executor 是产生于 worker 进程内部的线程，会执行同一个组件的一个或者多个 task。
3. Task: 实际的数据处理由 task 完成。

基于这样的架构设计，Storm 的工作流程如下图所示：

• 所有 Topology 任务的提交必须在 Storm 客户端节点上进行，提交后，由 Nimbus 节点分配给其他 Supervisor 节点进行处理。
• Nimbus 节点首先将提交的 Topology 进行分片，分成一个个 Task，分配给相应的 Supervisor，并将 Task 和 Supervisor 相关的信息提交到 Zookeeper 集群上。
• Supervisor 会去 Zookeeper 集群上认领自己的 Task，通知自己的 Worker 进程进行 Task 的处理。