当前访客身份:游客 [ 登录  | 注册加入尚学堂]
直播

我来了!

拥有积分:3868
尚学堂雄起!!威武。。。

博客分类

笔记中心

课题中心

提问中心

答题中心

解答题中心

Storm常见模式——分布式RPC

我来了! 发表于 2年前 (2014-10-28 22:22:03)  |  评论(0)  |  阅读次数(473)| 0 人收藏此文章,   我要收藏   

本文翻译自: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的整体工作过程如下:

  1. 接收到一个RPC调用请求;
  2. 发送请求到Storm上的拓扑;
  3. 从Storm上接收计算结果;
  4. 将计算结果返回给客户端。

以上过程,在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内部工作流程如下:

  1. Client向DRPC Server发送被调用执行的DRPC函数名称及参数。
  2. Storm上的topology通过DRPCSpout实现这一函数,从DPRC Server接收到函数调用流;
  3. DRPC Server会为每次函数调用生成唯一的id;
  4. Storm上运行的topology开始计算结果,最后通过一个ReturnResults的Bolt连接到DRPC Server,发送指定id的计算结果;
  5. DRPC Server通过使用之前为每个函数调用生成的id,将结果关联到对应的发起调用的client,将计算结果返回给client。

LinearDRPCTopologyBuilder

Storm提供了一个topology builder——LinearDRPCTopologyBuilder,它可以自动完成几乎所有的DRPC步骤。包括:

  1. 构建spout
  2. DRPC Server返回结果;
  3. 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也一样很简单。只需完成以下步骤:

  1. 启动DRPC Server(s);
  2. 配置DRPC Server(s)地址;
  3. 向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暴露给的不同人的个数。

为了完成这一计算,需要完成以下步骤:

  1. 获取所有点选了(tweet)该URL的人;
  2. 获取步骤1中所有人的关注者(followers,粉丝);
  3. 对所有关注者followers进行去重;
  4. 对步骤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"));
拓扑的执行分为以下四步:
  1. GetTweeters:获取所有tweet了指定URL的用户列表,这个Bolt将输入流[id, url]转换成输出流[id, tweeter],每个url元组被映射为多个tweeter元组。
  2. GetFollowers:获取步骤1中所有用户列表的followers,这个Bolt将输入流[id, twetter]转换成输出流[id, follower],当某个人同时是多个人的关注者follower,而且这些人都tweet了指定的URL,那么将产生重复的follower元组。
  3. PartialUniquer:将所有followers按照follower id分组,使得同一个follower在同一个task中被处理。这个Bolt接收follower并进行去重计数。
  4. 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

 

分享到:0
关注微信,跟着我们扩展技术视野。每天推送IT新技术文章,每周聚焦一门新技术。微信二维码如下:
微信公众账号:尚学堂(微信号:bjsxt-java)
北京总部地址:北京市海淀区西三旗桥东建材城西路85号神州科技园B座三层尚学堂 咨询电话:400-009-1906 010-56233821
Copyright 2007-2015 北京尚学堂科技有限公司 京ICP备13018289号-1 京公网安备11010802015183