知识点
相关文章
更多最近更新
更多Storm常见模式——分布式RPC
2019-03-02 23:44|来源: 网路
本文翻译自:https://github.com/nathanmarz/storm/wiki/Distributed-RPC,作为学习Storm DRPC的资料,转载必须以超链接形式标明文章原始出处及本文翻译链接。
分布式RPC(distributed RPC,DRPC)用于对Storm上大量的函数调用进行并行计算过程。对于每一次函数调用,Storm集群上运行的拓扑接收调用函数的参数信息作为输入流,并将计算结果作为输出流发射出去。
DRPC本身算不上Storm的特性,它是通过Storm的基本元素:streams,spouts,bolts,topologies而衍生的一个模式。DRPC可以单独作为一个独立于Storm的库发布,但由于其重要性还是和Storm捆绑在了一起。
总体概述
DRPC通过DRPC Server来实现,DRPC Server的整体工作过程如下:
- 接收到一个RPC调用请求;
- 发送请求到Storm上的拓扑;
- 从Storm上接收计算结果;
- 将计算结果返回给客户端。
以上过程,在client客户端看来,一个DRPC调用看起来和一般的RPC调用没什么区别。下面代码是client通过DRPC调用“reach”函数,参数为“http://twitter.com”:
DRPCClient client = new DRPCClient("drpc-host", 3772); String result = client.execute("reach", "http://twitter.com");
DRPC内部工作流程如下:
- Client向DRPC Server发送被调用执行的DRPC函数名称及参数。
- Storm上的topology通过DRPCSpout实现这一函数,从DPRC Server接收到函数调用流;
- DRPC Server会为每次函数调用生成唯一的id;
- Storm上运行的topology开始计算结果,最后通过一个ReturnResults的Bolt连接到DRPC Server,发送指定id的计算结果;
- DRPC Server通过使用之前为每个函数调用生成的id,将结果关联到对应的发起调用的client,将计算结果返回给client。
LinearDRPCTopologyBuilder
Storm提供了一个topology builder——LinearDRPCTopologyBuilder,它可以自动完成几乎所有的DRPC步骤。包括:
- 构建spout;
- 向DRPC Server返回结果;
- 为Bolt提供函数用于对tuples进行聚集。
下面是一个简单的例子,这个DRPC拓扑只是简单的在输入参数后追加“!”后返回:
public static class ExclaimBolt extends BaseBasicBolt { public void execute(Tuple tuple, BasicOutputCollector collector) { String input = tuple.getString(1); collector.emit(new Values(tuple.getValue(0), input + "!")); } public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("id", "result")); } } public static void main(String[] args) throws Exception { LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder("exclamation"); builder.addBolt(new ExclaimBolt(), 3); // ... }
由上述例子可见,我们只需很少的工作即可完成拓扑。当创建LinearDRPCTopologyBuilder
的时候,需要指定拓扑中
DRPC
函数的名称
“
exclamation”
。一个
DRPC Server
可以协调多个函数,每个函数有不同的函数名称。拓扑中的第一个
bolt
的输入是
两
个字段:第一个是请求的
id
号;第二个是请求的参数。
LinearDRPCTopologyBuilder
同时需要最后一个
bolt
发射一个包含两个字段的输出流:第一个字段是请求
id
;第二个字段是计算结果。因此,所有的中间
tuples
必须包含请求
id
作为第一个字段。
例子中,
ExclaimBolt
在输入
tuple
的第二个字段后面追加
“!”
,
LinearDRPCTopologyBuilder
负责处理其余的协调工作:与
DRPC Server
建立连接,发送结果给
DRPC Server
。
本地模式DRPC
DRPC可以以本地模式运行,下面的代码是如何在本地模式运行上面的例子:
LocalDRPC drpc = new LocalDRPC(); LocalCluster cluster = new LocalCluster(); cluster.submitTopology("drpc-demo", conf, builder.createLocalTopology(drpc)); System.out.println("Results for 'hello':" + drpc.execute("exclamation", "hello")); cluster.shutdown(); drpc.shutdown();
首先创建一个LocalDRPC
对象,该对象在本地模拟一个
DRPC Server
,正如
LocalCluster
在本地模拟一个
Storm
集群一样。然后创建一个
LocalCluster
对象在本地模式下运行拓扑。
LinearDRPCTopologyBuilder
含有单独的方法用于创建本地拓扑和远程拓扑。
本地模式下,
LocalDRPC
并不绑定任何端口,因此
Storm
的拓扑需要了解要通讯的对象
——
这就是为什么
createLocalTopology
方法需要以
LocalDRPC
对象作为输入。
加载完拓扑之后,通过对
LocalDRPC
调用
execute
方法,就可以执行
DRPC
函数调用了。
远程模式DRPC
在实际的Storm集群上运行DRPC也一样很简单。只需完成以下步骤:
- 启动DRPC Server(s);
- 配置DRPC Server(s)地址;
- 向Storm集群提交DRPC拓扑。
首先,通过storm脚本启动DRPC Server:
bin/storm drpc
然后,在Storm集群中配置DRPC Server地址,这就是DRPCSpout
读取函数调用请求的地方。这一步的配置可以通过
storm.yaml
文件或者拓扑的配置来完成。通过
storm.yaml
文件的配置方式如下:
drpc.servers:
- "drpc1.foo.com"
- "drpc2.foo.com"
最后,通过StormSubmitter
启动
DRPC
拓扑。为了以远程模式运行
上面的例子,代码如下:
StormSubmitter.submitTopology("exclamation-drpc", conf, builder.createRemoteTopology());
createRemoteTopology
被用于为
Storm
集群创建合适的拓扑。
一个复杂的例子
上面的exclamation只是一个简单的DRPC例子。下面通过一个复杂的例子介绍如何在Storm集群内进行DRPC——计算Twitter上每个URL的到达度(reach),也就是每个URL暴露给的不同人的个数。
为了完成这一计算,需要完成以下步骤:
- 获取所有点选了(tweet)该URL的人;
- 获取步骤1中所有人的关注者(followers,粉丝);
- 对所有关注者followers进行去重;
- 对步骤3中的关注者人数进行求和。
一个简单的URL到达度计算可能涉及成千上万次数据库调用以及数以百万的followers记录,计算量非常大。有了Storm,将很容易实现这一计算过程。单机上可能需要运行几分钟才能完成,在Storm集群上,即使是最难计算的URL也只需要几秒钟。
这个例子的代码在storm-starter:点击这里。这里是如何创建拓扑的代码:
LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder("reach"); builder.addBolt(new GetTweeters(), 3); builder.addBolt(new GetFollowers(), 12) .shuffleGrouping(); builder.addBolt(new PartialUniquer(), 6) .fieldsGrouping(new Fields("id", "follower")); builder.addBolt(new CountAggregator(), 2) .fieldsGrouping(new Fields("id"));
拓扑的执行分为以下四步:
- GetTweeters:获取所有tweet了指定URL的用户列表,这个Bolt将输入流[id, url]转换成输出流[id, tweeter],每个url元组被映射为多个tweeter元组。
- GetFollowers:获取步骤1中所有用户列表的followers,这个Bolt将输入流[id, twetter]转换成输出流[id, follower],当某个人同时是多个人的关注者follower,而且这些人都tweet了指定的URL,那么将产生重复的follower元组。
- PartialUniquer:将所有followers按照follower id分组,使得同一个follower在同一个task中被处理。这个Bolt接收follower并进行去重计数。
- CountAggregator:从各个PartialUniquer中接收各部分的计数结果,累加后完成到达度计算。
下面是PartialUniquer
这个
Bolt
的代码实现:
public class PartialUniquer extends BaseBatchBolt { BatchOutputCollector _collector; Object _id; Set<String> _followers = new HashSet<String>(); @Override public void prepare(Map conf, TopologyContext context, BatchOutputCollector collector, Object id) { _collector = collector; _id = id; } @Override public void execute(Tuple tuple) { _followers.add(tuple.getString(1)); } @Override public void finishBatch() { _collector.emit(new Values(_id, _followers.size())); } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("id", "partial-count")); } }
PartialUniquer
通过继承
BaseBatchBolt
实现了
IBatchBolt
接口,
batch bolt
提供了
API
用于将一批
tuples
作为整体来处理。每个请求
id
会创建一个新的
batch bolt
实例,同时
Storm
负责这些实例的清理工作。
当PartialUniquer
接收到一个
follower
元组时执行
execute
方法,将
follower
添加到请求
id
对应的
HashSet
集合中。
Batch bolt同时提供了finishBatch
方法用于当这个
task
已经处理完所有的元组时调用。
PartialUniquer
发射一个包含当前
task
所处理的
follower ids
子集去重后个数的元组。
在内部实现上,
CoordinatedBolt
用于检测指定的
bolt
是否已经收到指定请求
id
的所有
tuples
元组。
CoordinatedBolt
使用
direct streams管理实现这一协作过程。
拓扑的其他部分易于理解。到达度的每一步的计算过程都是并行进行的,通过DRPC实现也是非常容易的。
Non-linear DRPC拓扑
LinearDRPCTopologyBuilder
只能处理
“
线性的
”DRPC
拓扑
——
正如到达度这样可以通过一系列步骤序列来完成的计算。不难想象,
DRPC
调用中包含有更复杂的带有分支和合并
Bolt
的拓扑。目前,必须自己直接使用
CoordinatedBolt
来完成这种非线性拓扑的计算。
LinearDRPCTopologyBuilder工作过程
- DRPCSpout发射[args, return-info],其中return-info包含DRPC Server的主机和端口号,以及DRPC Server为该次请求生成的唯一id号;
- 构造一个Storm拓扑包含以下部分:
- DRPCSpout
- PrepareRequest(生成一个请求id,为return info创建一个流,为args创建一个流)
- CoordinatedBolt wrappers以及direct groupings
- JoinResult(将结果与return info拼接起来)
- ReturnResult(连接到DRPC Server,返回结果)
- LinearDRPCTopologyBuilder是建立在Storm基本元素之上的高层抽象。
高级进阶
- KeyedFairBolt用于组织同一时刻多请求的处理过程;
- 如何直接使用
CoordinatedBolt
。
转自:http://www.cnblogs.com/panfeng412/archive/2012/07/02/storm-common-patterns-of-distributed-rpc
相关问答
更多-
分布式系统的例子和这个例子的工作模式[2022-03-11]
分布式软件系统(Distributed Software Systems)是支持分布式处理的软件系统,是在由通信网络互联的多处理机体系结构上执行任务的系统。它包括分布式操作系统、分布式程序设计语言及其编译(解释)系统、分布式文件系统和分布式数据库系统等。 分布式操作系统负责管理分布式处理系统资源和控制分布式程序运行。它和集中式操作系统的区别在于资源管理、进程通信和系统结构等方面。 分布式程序设计语言用于编写运行于分布式计算机系统上的分布式程序。一个分布式程序由若干个可以独立执行的程序模块组成,它们分布于一个 ... -
分布式系统的例子和这个例子的工作模式[2023-05-31]
分布式软件系统(Distributed Software Systems)是支持分布式处理的软件系统,是在由通信网络互联的多处理机体系结构上执行任务的系统。它包括分布式操作系统、分布式程序设计语言及其编译(解释)系统、分布式文件系统和分布式数据库系统等。 分布式操作系统负责管理分布式处理系统资源和控制分布式程序运行。它和集中式操作系统的区别在于资源管理、进程通信和系统结构等方面。 分布式程序设计语言用于编写运行于分布式计算机系统上的分布式程序。一个分布式程序由若干个可以独立执行的程序模块组成,它们分布于一个 ... -
分布式系统的例子和这个例子的工作模式[2022-07-30]
分布式软件系统(Distributed Software Systems)是支持分布式处理的软件系统,是在由通信网络互联的多处理机体系结构上执行任务的系统。它包括分布式操作系统、分布式程序设计语言及其编译(解释)系统、分布式文件系统和分布式数据库系统等。 分布式操作系统负责管理分布式处理系统资源和控制分布式程序运行。它和集中式操作系统的区别在于资源管理、进程通信和系统结构等方面。 分布式程序设计语言用于编写运行于分布式计算机系统上的分布式程序。一个分布式程序由若干个可以独立执行的程序模块组成,它们分布于一个 ... -
duboo其实是阿里巴巴做的,服务和营销都是好的,他俩其实没什么大区别,只不过dubbo 可以 和 Spring框架无缝 集成。
-
软件开发中经常提到一个分布式这个词,请问什么是分布式开发[2023-11-11]
现在的软件开发都讲究个"层"的意思. 分布式开发将一个系统分为三个层次:客户端应用程序,应用程序服务器,后台数据库。客户端提出请求,应用服务器接受请求并处理然后返回数据给客户端,后台数据库当然是提供数据。多半是用于WEB开发.这样的分层开发有很多 好处..我就不多说了... -
分布式到底是什么意思[2023-07-24]
这个比较复杂,这个属于架构方面的,大概是指客户端和服务器端的关系。以前的程序的服务端比较集中在一块,分布式的服务器端可能分布在不同的地方,如云端等等。。。 -
什么是分布式系统?[2022-11-11]
一、DFS为何物? DFS 即微软分布式文件系统的简称,系统管理员可以利用它来有效的整合网络资源,并把这些资源以单一的层次结构呈现给网络用户。管理员利用它可以把资源发布成一 个树形结构,这样大大简化了为用户进行资源配置和对资源管理的工作量。我们可以在不同的机器上调整和移动文件,这不会影响到用户的访问。 二、为什么要使用DES? 1、DFS使用了现有网络中的Share权限,管理员不必进行新的配置 2、通过一个DFS树形结构用户就可以访问多个网络资源,而不用再把远程驱动器映射到本地共享资源中。 3、DFS可以配 ... -
什么是分布式系统?[2024-03-23]
分布式系统(distributed system)是建立在网络之上的软件系统。正是因为软件的特性,所以分布式系统具有高度的内聚性和透明性。因此,网络和分布式系统之间的区别更多的在于高层软件(特别是操作系统),而不是硬件。内聚性是指每一个数据库分布节点高度自治,有本地的数据库管理系统。透明性是指每一个数据库分布节点对用户的应用来说都是透明的,看不出是本地还是远程。在分布式数据库系统中,用户感觉不到数据是分布的,即用户不须知道关系是否分割、有无复本、数据存于哪个站点以及事务在哪个站点上执行等。 故名思义,分布式 ... -
只需按照群集设置指南操作: https://storm.apache.org/documentation/Setting-up-a-Storm-cluster.html 对于伪分布式设置,请在单台计算机中运行所有守护程序(ZK,Nimbus和一个单独的主管)。 Just follow the cluster setup guide: https://storm.apache.org/documentation/Setting-up-a-Storm-cluster.html For pseudo distri ...
-
除非出现问题,否则你已经完成了。 当您向Storm提交拓扑时,Nimbus服务会通过遍布整个群集的Supervisor进程查看群集上的负载。 然后,Nimbus为拓扑运行提供一定数量的资源。 这些资源通常遍布集群中的各个Supervisor节点,并且它们将并行处理元组。 Nimbus偶尔会重新审视这些决策并更改哪些节点处理什么以尝试保持群集中的负载平衡。 作为用户,您永远不应该注意到这个过程。 假设您的Storm集群设置正确,您唯一需要做的就是提交拓扑。 Storm为您处理整个多节点并行处理事务。 也就是说 ...