大家好，我们今天主要讨论以下几个问题：
机器学习与实时处理系统应用现在我们先来看看第一部分：机器学习与实时处理系统应用。我们首先简单了解下机器学习，然后引入分布式实时处理系统的概念以及实时处理系统与机器学的关系。 机器学习在现实世界中的作用越来越重要。 机器学习的方法非常多，比如传统的知识库方法，类比方法，归纳方法，演绎方法等各种方法。 目前在大多数领域中应用最多的当属归纳学习方法。 在通常的归纳型机器学习中，我们的目标是让计算机学习到一个“模型”（这种模型是人类预先组织好的，有固定的数据结构和算法等等），然后我们就可以用这个“模型”来进行“预测”。 预测就是从现实中输入一些数据，通过学习到的模型进行计算，得到的输出。我们希望这个模型可以在很高的概率下输出一个和真实结果差距不大的结果。 一旦我们得到了这个模型，我们可以使用该模型处理输入数据，得到输出数据（即预测结果），而归纳性机器学习的任务就是学习中间的这个模型。 如果我们将这个模型看成一个函数，那么我们可以认为归纳性机器学习的目的就是学习得到一个函数F，如果该函数的参数为x，输出为y。那么我们希望学到的东西就是 y = F(x) 中的F。 我们先用一个最简单的例子来讲一下： 假设我们现在不知道一个物体自由落体速度的计算公式，需要学习如何预测一个物体的自由落体速度，机器学习的第一步就是收集数据，假设我们可以测量出物体下坠的任何时间点的速度，那么我们需要收集的数据就是某个物体的下坠时间和那个时间点的速度。 现在我们收集到一系列数据：
我们这里给出两个假设。第一个假设是，一个物体自由落体的速度只和时间有关系 第二个假设是，我们可以使用一个简单的“模型”：一元一次函数得到物体的速度。（即 F(x) = ax b） 在这个问题中，a、b 这就是这个模型待学习的“参数”。 现在的问题就是我们需要用什么策略来学习这些参数？因为我们可以遍历的数值空间是无穷大的，因此我们必须采用某种策略指导我们进行学习。我们就用非常朴素的思想来将解决这个问题吧。 在正式学习前，我们先将收集的数据分成两组，一组是“训练数据”，一组是“测试数据” 。 假设训练数据是：
测试数据是：
我们需要根据训练数据计算出我们的参数a和b。 然后使用我们计算出来的a和b预测测试数据，比较F(x)和实际数据的差距，如果误差小到一定程度，说明我们学习到的参数是正确的，比如和实际数据的差距都小于5% 。 如果满足条件说明参数正确，否则说明参数不够精确，需要进一步学习，这个差距，我们称之为误差（Loss）。现在我们来看一下在这个模型（简单的一元一次线性函数）下如何学习这两个参数 比如我们可以采用这种学习策略 1.首先a和b都假定为整数，假定a的范围是[-10, 10]这个区间，b的范围是[-100, 100]这个区间 2.遍历所有的a和b的组合，使用a和b计算ax b，x取每个训练数据的输入数据，评估计算结果精确性的方法是计算结果和训练数据结果的差的绝对值除以训练数据结果，也就是 Loss = |F(x) - Y| / Y 3.计算每个组合的Loss的平均值，取平均Loss最小的为我们假定的“学习结果” 。 现在我们就得到了a和b，并且这个a和b是在我们给定范围里精度最高的参数，我们用这个a和b去训练数据里面计算平均的 |F(x) - Y| / Y，如果平均Loss小于 5%，说明这个a和b是符合我们精度的。否则我们需要优化我们的学习策略。 这种朴素的基于归纳学习的机器学习方法可以分为以下几步：
这里我们也要注意，上述步骤的前提是我们的模型是可以收敛的，如果模型本身就是发散的，那么我们就永远得不到我们的结果了。 机器学习与实时处理系统传统的机器学习是一种批处理式的方法，在这种方法下，我们需要预先准备好所有的训练数据，对训练数据进行精心组织和筛选，很多情况下还需要对数据进行标记（监督式学习），而训练数据的组织会对最后的训练结果产生相当大的影响。 在这种算法中我们要处理完所有数据后才能更新权重和模型。 但现在出现了许多在线学习算法，这种算法可以对实时输入的数据进行计算，马上完成权重和模型更新。 一方面我们可以用于监督式学习（完成数据标记后马上加入训练），也可以用于大量数据的非监督式学习。 而在这种情况下，实时处理系统就可以大展身手了。在线系统和实时处理系统可以确保实时完成对数据的学习，利用实时新系统。 实现思路如下图所示： 这里我们可以看到，系统接收来自其他系统的实时输入，然后实时处理系统中使用在线算法快速处理数据，实时地更新模型权重信息。 纯粹的在线算法可能并不适合许多情景，但是如果将部分在线算法和传统的批处理式算法结合，将会起到非常好的效果。而且许多数据分析工作确实可以通过这种方式完成一部分处理，至少是预处理。 目前机器学习的趋势就是对精度和速度的要求越来越高，方法越来越复杂，而数据越来越多，计算量越来越大，如果没有足够的计算资源，不一定能够在有限时间内完成足够的学习，因此现在类似于Tensorflow之类的机器学习解决方案都会提供针对分布式的支持。而大数据场景下的机器学习也变得越来越重要，这也对我们的分布式计算与存储方案提出了严峻的挑战。 分布式计算拓扑搭建现在我们来看一个现实工程中常常会遇到的问题。 我们在开发实际系统时常常会收集大量的用户体验信息，而我们常常需要对这些体验信息进行筛选、处理和分析。那么我们应该如何搭建一个用于实时处理体验信息的分布式系统呢？ 我们先来看一下整体流程：
业务系统调用体验信息接口，将体验信息信息异步写入到特定的文件当中。使用永不停息的体验信息检测程序不断将新生成的体验信息发送到数据处理服务器。
首先数据处理服务器的体验信息接收负责将体验信息写入本地的Redis数据库中。然后我们使用消息源从Redis中读取数据，再将数据发送到之后的消息处理单元，由不同的数据处理单元对体验信息进行不同处理。
消息处理单元完成体验信息处理之后，将体验信息处理结果写入到Cassandra数据库中，并将体验信息数据写入到Elasticsearch数据库中。 其中关键的部分就是图中用长方形框出来的部分，该部分的作用是完成对数据的筛选、处理和基本分析。这部分我们将其称作计算拓扑，也就是用于完成实际计算的部分。 我们接下来阐述一下每一步具体如何做。 收集体验信息收集体验信息分为以下几步：
处理体验信息接下来是处理体验信息，处理体验信息主要在计算拓扑中完成。分为四步：
存储结果最后就是对计算结果的存储，我们需要使用存储模块将数据写入到不同的数据库中：
我们可以发现，在上面几步中，其他都可以使用现成的系统来完成任务，最关键的部分就是计算拓扑，计算拓扑需要高实时性地完成体验信息处理分析任务，这样才能应付大型系统中以极快速度产生的大量体验信息。 这里我们可以使用一个独立的计算集群来完成这个事情。每个计算节点负责完成一个计算任务，完成之后将数据传送给下一个计算节点完成后续的计算任务。每个计算节点都有一个消息队列用于接收来自上一级的消息，然后处理消息并继续将结果发送给下一级的计算节点。 消息处理的问题这里我们通常关心三个问题：
数据完全处理 我们先来看一下如何解决解决数据的完全处理问题。 我们这里讲每一个需要处理的数据（一条体验信息记录）组织成一个Tuple，也就是元组。每个计算节点都以Tuple为单位进行数据处理。每个元组都会有一个ack方法，用于告知上一级计算节点该Tuple已经处理完成。 我们以下面的方式处理Tuple，保证所有数据都会被完全处理：
数据流量控制 第二个需要解决的问题就是数据流量控制问题。 我们可以设想一下，如果网络状况不好，在特定时间内有许多元组都没有得到处理，那么数据源节点就会重发许多Tuple，然后后续节点继续进行处理，产生更多的Tuple，加上我们需要正常处理的Tuple，使得集群中的Tuple越来越多。而由于网络状况不好，节点计算速度有限，会导致集群中积累的过多数据拖慢整个集群的计算速度，进一步导致更多的Tuple可能计算失败。 为了解决这个问题，我们必须想方设法控制集群中的流量。 这个时候我们就会采用一种流量背压机制。该机制借鉴自Twitter Heron。 拓扑结构设计最后就是如何搭建这个拓扑，并尽量高效地完成计算了。 在分布式实时处理系统领域，目前最为成功的例子就是Apache Storm项目，而Apache Storm采用的就是一种流模型。而我们的Hurricane则借鉴了Storm的结构，并进行了简化（主要在任务和线程模型上）。 这种流模型包括以下几个概念：
其中有以下几个组件：
从任务的抽象角度来讲，每个Executor之间会相互传递数据，只不过都需要通过Manager完成数据的传递，Manager会帮助Executor将数据以元组的形式传递给其他的Executor。 Manager之间可以自己传递数据（如果分组策略是确定的），有些情况下还需要通过President来得知自己应该将数据发送到哪个节点中。 了解整体架构后，我们来具体讲解一下President和Manager的架构。 President的架构如图所示： President的底层是一个基于Meshy实现的NetListener，该类负责监听网络，并将请求发送给事件队列，交由事件队列处理。 President的核心是EventQueue。这是一个事件队列，当没有计算任务的时候，会从事件队列中获取事件并进行处理。 用户需要在EventQueue中事先注册每个事件对应的处理函数，President会根据事件类型调用对应的事件处理函数。 接下来是Manager的架构。Manager的架构相对来说较为复杂。考虑到性能优化等问题，这个架构修改了几次。 首先，最顶层和President一样，是一个事件队列，并使用一个基于Meshy的NetListener来完成IO事件的响应（转换成事件放入事件队列）。 接下来有两个模块：
再下一层就是Executor。Executor分为SpoutExecutor和BoltExecutor，每个Executor都是一个单独的线程，在系统初始化Topology的时候，Managert会初始化Executor，并设置其中的任务。SpoutExecutor负责执行Spout任务，而BoltExecutor负责执行Bolt任务。 其中BoltExecutor需要接受来自其他Executor的Tuple，因此包含一个Tuple Queue。Tuple Dispatcher会将Tuple投送到这个Tuple Queue中，而Bolt则从Tuple Queue中取出数据并执行任务。 Eexecutor在执行完任务后，可能会将Tuple通过OutputCollector投送到OutputQueue中。我们又设计了一个OutputDispatcher，从OutputQueue中获取Tuple并发送到其他节点。OutputQueue也是一个带锁的阻塞队列，是唯一用于输出的队列。 现在我们来详细介绍一下Hurricane的基本组件。 Task Task是对计算任务的统一抽象，规定了计算任务的统一接口。Spout和Bolt都是Task的特殊实现。 Task包含三个接口函数：
Spout Spout是Task的特例，任务用于产生待处理的元组。因此除了Task的接口以外，还增加了两个新接口。
这里需要注意一下，Hurricane会有简单的背压机制，当Bolt检测到Tuple流量过大的时候，会向Spout进行反馈，Spout会随之降低其发送元组的速度。 如果Bolt处理速度大于Tuple的生成速度，Bolt又会向Spout反馈增加流量，Spout会放松流量限制。 Bolt Bolt是计算单元，负责处理来自其他Spout和Bolt的元组。Bolt同样是特殊的Task，因此除了Task接口外，还有两个新的接口。
Bolt接收元组的方式我们称之为分组方式。目前hurricane支持3种分组方式。
Executor Executor负责执行具体的任务。每个Executor是一个独立的线程，可以充分利用多核和多线程的CPU。为了简化模型，每个Executor只负责执行一个任务。
如果TupleQueue中不包含元组，BoltExecutor会被阻塞。超过一定时间没有获取到元组，BoltExecutor会向Spout反馈，解除部分流量限制，加快元组生成速度。 其他特性Squared 介绍完Hurricane的基本功能与架构之后，我们来介绍一下Squared。 首先我们解释一下Squared是什么？ 左侧是Hurricane基本的计算模型，在该计算模型中，系统是一个计算任务组成的网络。我们需要考虑每个节点的琐屑实现。 但如果在日常任务中，使用这种模型相对来说会显得比较复杂，尤其当网络非常复杂的时候。 为了解决这个问题，看一下右边这个计算模型，这是对我们完成计算任务的再次抽象。
这里其实是将网络映射成了简单的数据操作流程。这样一来，解决问题和讨论问题都会变得更为简单直观。 这就是Squared所做的事情将基于网络与数据流的模型转换成这种简单的流模型，让开发者更关注于数据的统计分析，脱离部分繁琐的工作。 保序 在现实的工作中，我们常常需要一个的特性就是保序。 比如部分银行交易和部分电商订单处理，希望数据按照顺序进行处理，但是传统的数据处理系统往往不支持这个特性。所以我们就实现了保序功能。 保序的实现原理很简单，首先每个Tuple会一个一个orderId字段，orderId是依据顺序生成的，然后所有对Tuple的操作都会检验该orderId之前的Tuple是否已经完成。 声明：文章版权归原作者所有 部分文章转自互联网 如有侵权请联系 [邮箱地址] 删除 |
注册
登录
