Storm高级使用模式

DRPC原理

1. 什么是RPC

a. RPC 功能目标

RPC 的主要功能目标是让构建分布式计算（应用）更容易，在提供强大的远程调用能力时不损失本地调用的语义简洁性。为实现该目标，RPC 框架需提供一种透明调用机制让使用者不必显式的区分本地调用和远程调用。 下面我们将具体细化 stub 结构的实现。

b. RPC 调用分类

RPC 调用分以下两种：

同步调用 : 客户方等待调用执行完成并返回结果。

异步调用 : 客户方调用后不用等待执行结果返回，但依然可以通过回调通知等方式获取返回结果。 若客户方不关心调用返回结果，则变成单向异步调用，单向调用不用返回结果。

异步和同步的区分在于是否等待服务端执行完成并返回结果。

2. Storm DRPC

Storm是一个分布式实时处理框架,它支持以DRPC方式调用.可以理解为Storm是一个集群,DRPC提供了集群中处理功能的访问接口.其实即使不通过DRPC，而是通过在Topoloye中的spout中建立一个TCP/HTTP监听来接收数据，在最后一个Bolt中将数据发送到指定位置也是可以的。这是后话，后面再进行介绍。而DPRC则是Storm提供的一套开发组建，使用DRPC可以极大的简化这一过程。

Storm里面引入DRPC主要是利用storm的实时计算能力来并行化CPU intensive的计算。DRPC的storm topology以函数的参数流作为输入，而把这些函数调用的返回值作为topology的输出流。

DRPC其实不能算是storm本身的一个特性， 它是通过组合storm的原语spout，bolt， topology而成的一种模式(pattern)。本来应该把DRPC单独打成一个包的， 但是DRPC实在是太有用了，所以我们把它和storm绑定在一起。

3. DRPC Server

a. 接收一个RPC请求。

b. 发送请求到storm topology

c. 从storm topology接收结果。

d. 把结果发回给等待的客户端。

从客户端的角度来看一个DRPC调用跟一个普通的RPC调用没有任何区别。比如下面是客户端如何调用RPC： reach方法的,方法的参数是:

http://twitter.com。

DRPCClient client = new DRPCClient("drpc-host", 3772);

String result = client.execute("reach","http://twitter.com");

DRPC的工作流程：

客户端给DRPC服务器发送要执行的方法的名字，以及这个方法的参数。实现了这个函数的topology使用DRPCSpout从DRPC服务器接收函数调用流。每个函数调用被DRPC服务器标记了一个唯一的id。

这个topology然后计算结果，在topology的最后一个叫做ReturnResults的bolt会连接到DRPC服务器，并且把这个调用的结果发送给DRPC服务器(通过那个唯一的id标识)。DRPC服务器用那个唯一id来跟等待的客户端匹配上，唤醒这个客户端并且把结果发送给它。

DRPC构建与使用

1. 本地DRPC

DRPC可以以本地模式运行， 主要用来调试。下面就是以本地模式运行 上面例子的代码:

LocalDRPC drpc = newLocalDRPC();

LocalCluster cluster = newLocalCluster();

cluster.submitTopology(

    "drpc-demo",

    conf,

    builder.createLocalTopology(drpc)

);

cluster.shutdown();

drpc.shutdown();

首先你创建一个LocalDRPC对象。 这个对象在进程内模拟一个DRPC服务器，跟LocalClusterLinearTopologyBuilder有单独的方法来创建本地的topology和远程的topology。在本地模式里面LocalDRPC对象不和任何端口绑定，所以我们的topology对象需要知道和谁交互。这就是为什么createLocalTopology方法接受一个LocalDRPC对象作为输入的原因。

把topology启动了之后，你就可以通过调用LocalDRPC对象的execute来调用RPC方法了。

2. 远程DRPC

a. 启动DRPC Server

b. 配置DRPC服务器的地址

c. 提交DRPC Topology到Storm集群

3. LinearDRPCTopologyBuilder(自动完成几乎所有的DRPC步骤)：

a. 构建Spout

b. 向DRPC Server返回结果

c. 给Bolt提供有限聚合几组tuple的能力

Trident

Trident是对Storm的更高一层的抽象,除了提供一套简单易用的流数据处理API之外，它以batch(一组tuples)为单位进行处理，这样一来，可以使得一些处理更简单和高效。

我们知道把Bolt的运行状态仅仅保存在内存中是不可靠的，如果一个node挂掉，那么这个node上的任务就会被重新分配，但是之前的状态是无法恢复的。因此，比较聪明的方式就是把storm的计算状态信息持久化到database中，基于这一点，trident就变得尤为重要。因为在处理大数据时，我们在与database打交道时通常会采用批处理的方式来避免给它带来压力，而trident恰恰是以batch groups的形式处理数据，并提供了一些聚合功能的API。

Storm是一个实时流计算框架，Trident是对storm的一个更高层次的抽象，Trident最大的特点以batch的形式处理stream。

一些最基本的操作函数有Filter、Function，Filter可以过滤掉tuple，Function可以修改tuple内容，输出0或多个tuple，并能把新增的字段追加到tuple后面。

聚合有partitionAggregate和Aggregator接口。partitionAggregate对当前partition中的tuple进行聚合，它不是重定向操作。Aggregator有三个接口：CombinerAggregator, ReducerAggregator，Aggregator，它们属于重定向操作，它们会把stream重定向到一个partition中进行聚合操作。

重定向操作会改变数据流向，但不会改变数据内容，重定向操会产生网络传输，可能影响一部分效率。而Filter、Function、partitionAggregate则属于本地操作，不会产生网络传输。

GroupBy会根据指定字段，把整个stream切分成一个个grouped stream，如果在grouped stream上做聚合操作，那么聚合就会发生在这些grouped stream上而不是整个batch。如果groupBy后面跟的是aggregator，则是聚合操作，如果跟的是partitionAggregate，则不是聚合操作。

上面的例子都可以在github上找到： https://github.com/pereferrera/trident-hackaton/

多语言开发

storm由closure实现，但通过 multilang protocl （多语言协议），能够使用php，python，ruby或者javascript来写spout和bolt。 多语言协议是storm中实现的一种特殊协议，它使用标准输入和标准输出作为与执行spout和bolt任务的进程之间通信的信道。消息以json格式或者普通的文本行通过信道传输。

Topology的定义本身是一个Thrift结构，并且Nimbus就是一个Thrift服务， 你可以提交由任何语言创建的topology。