实时数据处理：云计算与大数据应用研究

数据的值随着时间的推移而减少，因此必须在事件发生时立即进行处理。最好在数据出现时立即处理它，并且发生单个事件，而不是分批缓存。这就是计算流量的原因。

实时搜索、高频交易、社交网络等新应用的出现，将传统的数据处理系统推向了极限。这些新的应用要求流计算解决方案是可扩展的，可以处理高频数据流和大规模数据。虽然Map Reduce等分布式批处理技术可以处理越来越多的数据，但这些技术并不适合于实时数据处理，也不能简单地将Map Reduce转换成一个实时计算框架。实时数据处理系统和批量数据处理系统在需求上有本质的区别，这主要体现在消息管理（数据传输）上。实时处理系统需要维护一个由消息队列和消息处理器组成的实时处理网络。消息处理器需要从消息队列中获取消息以进行处理，更新数据库，向其他队列发送消息，等等。主要体现在以下几个方面。

（1）消息处理逻辑代码的比例很小。它主要涉及消息框架的设计和管理，需要配置消息发送的位置，部署消息处理器，部署中间消息节点。

（2）健壮性和容错性：所有消息处理程序和消息队列都需要保证正常运行。

（3）可伸缩性：当消息处理程序达到阈值时，必须将数据分流，并配置新的处理程序来处理分流的消息。

对于分布式消息处理系统，最终的分解是消息队列和消息处理器的结合，而消息处理无疑是实时计算的基础。Twitter的Storm和Yahoo的S4是在这方面提出的解决方案。

Store是Back Type开发的分布式容错实时计算系统，它托管在Git Hub上，遵循Eclipse Public License 1.0。Storm为分布式实时计算提供了一组通用的原语，就像Map Reduce框架中的Map和Reduce一样，它可以用于“流处理”、实时处理消息和更新数据库。Storm的工程师Nathan Marz说：“Storm可以轻松地在一组计算机中编写和扩展复杂的实时计算。Storm是实时处理，而Hadoop是批处理。Storm保证每条消息都被处理，并且在一个能够每秒处理数百万条消息的小型集群中，它是快速的。更好的是，您可以使用任何编程语言进行开发。”

Storm的主要特征如下。

（1）简单规划模型。与Map Reduce类似，它降低了并行批处理的复杂性，而Storm降低了实时处理的复杂性。

（2）可以使用多种编程语言。可以在Storm上使用多种编程语言，默认情况下支持Clojure、Java、Ruby和Python。要增加对其他语言的支持，只需实现一个简单的Storm通信协议。

（3）Storm管理工作流程和节点中的故障。

（4）水平膨胀。计算在多个线程、进程和服务器之间并行执行。

（5）可靠的消息处理。Storm确保每条消息至少被完整地处理一次。当任务失败时，它负责重新尝试来自消息源的消息。

（6）速度快。该系统的设计是为了确保消息可以被快速处理，使用ZeroMQ作为它的底层消息队列。

（7）局部模式。Storm有一个“本机模式”，在处理过程中完全模拟Storm集群，允许快速开发和单元测试。

1.系统体系结构

Storm系统体系结构簇由一个主节点和多个工作节点组成。主节点运行一个名为“Nimbus”的守护进程来分配代码、分配任务和故障排除。每个Work节点运行一个名为“Supervisor”的守护进程，用于侦听工作、启动和终止工作流程。Nimbus和Supervisor都可以快速恢复并处于无状态，因此它们变得非常健壮，它们之间的协调是由Apache Zoo Keeper完成的。

2.工作原理

Storm术语包括消息流、消息源、消息处理程序、任务、工作流程、消息分发策略和拓扑。消息流系统是指正在处理的数据，消息源是源数据，消息处理程序是处理过的数据，任务是在消息源或消息处理程序中运行的线程，工作进程是运行这些线程的进程，消息策略指定消息处理程序作为输入数据接收到什么。

（1）计算拓扑

实时计算应用程序的逻辑封装在Storm中的拓扑对象中。除非用户显式终止Storm拓扑，否则该拓扑将始终运行。一个拓扑是消息源和消息处理程序的有向图，其中大多数是有向无圈图，而连接消息源和消息处理程序的是Stream lE。

（2）消息流

消息流是Storm中最关键的抽象，它是一个无限元组（Tuple）序列。消息流的定义主要是消息流中元组的定义，即元组中每个字段的定义（类似于数据库中的表和属性）。元组的字段类型可以是整数、长、短、字节、字符串、双、浮点、布尔和字节数组。

相应的序列化程序现在可用。

每个消息流都定义了一个ID。Output Fields Declarer定义了允许在不指定ID的情况下定义流的方法，在这种情况下，Stream将具有默认的ID。

（3）消息来源

消息源是拓扑中的消息生成器。通用消息源从外部源读取数据并向拓扑发送消息。源可以是可靠的，也可以是不可靠的。可靠的消息源可以重传消息，而不可靠的消息源不能重传消息。

消息源可以发出多个消息流，使用Out Fields Declarer.ramre Stream定义多个消息流，然后使用Spout Output Collection发送指定的消息流。

消息源类中最重要的方法是nexttuple，或者发送新消息，或者返回没有新消息的消息。请注意，nextTuple方法不能阻止消息源（块吐出）实现，因为Storm调用同一线程上所有消息源的方法。

另外两个更重要的源方法是ack和Failure。Storm通过ack和Failure保证拓扑可靠性（容错），在成功处理消息时调用ack标记数据处理（类似于断点）的过程，如果消息处理失败则调用失败恢复。

（4）消息处理程序

消息处理逻辑封装在消息处理器中，如过滤、聚合、查询数据库等。复杂的消息流处理通常需要经过许多步骤，即通过多步消息处理器。消息处理程序可以简单地传递消息流，或者他们可以发送多个消息流，使用Output Fields Deder-Randre Stream定义消息流，并使用Output Collection.emit选择要传输的消息流。

消息处理程序的主要方法是执行，它以消息作为输入，消息处理程序使用Output Collection来传输消息。消息处理程序必须为它处理的每一条消息调用Output Collection的方法。通知Storm消息处理已完成。一般过程是消息处理程序处理输入消息，发出零条或多条消息，然后调用ack通知Storm消息已被处理，Storm提供一个IBasicBolt调用来自动进行ack。

（5）消息分发策略

消息流组用于定义消息流应该分配给消息处理程序的多个任务。风暴中有六种类型的流群。

1）混乱组：在消息流中随机分发消息，确保每个消息处理程序接收的消息数量相同。(www.xing528.com)

2）字段组：按字段（如userid）分组，将具有相同userid的消息分配给相同的消息处理程序，而将不同的userid分配给不同的消息处理程序。

3）所有组：广播发送，对于每条消息，所有消息处理程序都将接收。

4）全局组：将消息分配给Storm中的一个消息处理程序的任务之一的全局分组，该任务分配给ID的最低值。

5）Non组：不分组，即消息流不关心到底谁会收到它的消息。目前这种分组和Shuffle组是一样的效果，有一点不同的是Storm会把这个消息处理者放到其订阅者的线程里面执行。

6）直接分组：直接分组，这是一种相对特殊的分组方法，它意味着消息的发送方指定消息由接收方的任务处理。只有声明为Direct Stream的消息流才能声明此分组方法，并且必须使用emit Direct方法发出消息。消息处理程序可以获得通过Topology Context处理其消息的Task ID。

（6）可靠性

Storm保证每项任务将在拓扑上全部完成。Storm跟踪每个消息源任务生成的任务树（另一条消息处理程序在处理一项任务后可以发送另一条消息，从而形成一棵树），跟踪该任务树，直到该树被成功处理为止。每个拓扑都有一个消息超时设置，如果Storm未能在此时间内检测到消息树的成功执行，则拓扑将消息标记为执行失败并重新发送消息。

为了利用Storm的可靠性，必须在发送新消息和处理消息时通知它。这是由Output Collection完成的。通过其发出方法生成新消息，并通过其ack方法通知消息处理。

（7）任务

每个消息源和消息处理程序都在整个集群中执行多个任务。每个任务对应于一个线程，而Stream组定义如何从一个任务触发消息到另一个任务。可以调用Topology Builder.set Spout和Top Builder.set Bolt来设置并行度以确定有多少任务。

（8）工作过程

拓扑可以在一个或多个工作过程中执行，每个过程执行整个拓扑的一部分。对于并行性为300的拓扑，如果使用50个辅助进程，则每个辅助进程将处理其中的6个任务。Storm尽可能均匀地为所有工作过程分配拓扑。

（一）S4分布式流计算平台

S4是雅虎发布的通用、可扩展、部分容错和插件式分布式流计算平台。在这个平台上，程序员可以轻松地开发处理流数据的应用程序。

雅虎开发S4的主要目的是处理用户反馈：在搜索引擎的广告中，用户点击的可能性是根据当前情况（用户偏好、地理位置、查询和点击）来估算的。

S4的设计目标如下。

（1）采用分散、对称的结构：无中心节点和特殊功能节点（易于部署和维护），提供简单的编程接口。

（2）设计了一个由通用硬件组成的高可用性、良好可扩展性的集群。

（3）尽量减少延迟：使用本地内存，尽量避免磁盘I/O。

（4）可插接结构，满足一般用户需求。

（5）设计思想应更友好：易于编程，更灵活。

但是，S4集群不允许添加或删除节点，在发生故障时允许数据丢失，并且不考虑系统的负载平衡和健壮性。

1.系统体系结构

S4提供客户端（客户端）和适配器（适配器），供第三方客户端访问S4集群，S4集群构成S4系统的三个组件，即客户机（客户端）、适配器（适配器）和简单的可伸缩流处理系统集群。这三个部分通过通信协议发送和接收消息，客户端与适配器之间的交互采用TCP/IP协议，适配器与S4集群之间的交互采用UDP协议。

为了使整个集群体系结构满足业务需求，S4体系结构的设计考虑了以下几点。

（1）S4系统体系结构的代理模式。为了在公共机器集群上处理分布式处理，并且在集群中没有共享内存，S4体系结构使用Actor模式，它提供封装和地址透明性语义。因此，在允许大规模并发的同时，它还提供了一个简单的编程接口。S4系统通过处理单元进行计算，并且消息以数据事件的形式在处理单元之间传输，PE消费事件发出一个或多个可由其他PE处理的事件，或直接发布结果。每个PE的状态对于其他PE是不可见的，PE之间唯一的交互方式是发出事件和使用事件。该框架提供了将事件路由到适当PE并创建新PE实例的能力。S4设计模式与封装和地址透明性兼容。

（2）集群的P2P点对点体系结构。为了简化部署和操作以获得更大的稳定性和可伸缩性，S4使用了对等结构，集群中的所有处理节点都相同，没有中央控制。该体系结构将使集群具有很强的可扩展性，在理论上可以处理节点总数无上限的问题，同时，S4不存在单一的容错问题。

（3）通用模块的可插拔特性。

（4）S4系统是用Java开发的，采用了非常丰富的模块化编程。每个通用功能点尽可能抽象为一个通用模块，并尽可能多地定制每个模块。

2.关键部件

（1）Client

S4中的所有事件流都由客户端触发。Client是S4提供的第三方客户端，它通过驱动程序组件与Adapter进行交互，并通过Adapter从S4集群接收或发送消息。

（2）适配器

适配器负责与S4集群交互，接收客户端发送到S4集群的请求，监听S4集群返回的数据并发送给客户端，使用TCP/IP协议提高客户端与适配器之间通信的可靠性。与S4集群的交互是基于UDP协议来提高传输速率的。

适配器也是一个集群，具有多个Adapter节点，客户端可以通过多个驱动程序与多个Adapter通信，这确保了当单个客户端分发大量数据时，Adapter不会成为瓶颈。它还确保系统支持多个客户端应用程序并发执行的速度、效率和可靠性。