Storm入门Twitter Storm: DRPC简介
Posted 静候东风
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Storm入门Twitter Storm: DRPC简介相关的知识,希望对你有一定的参考价值。
作者: xumingming | 可以转载, 但必须以超链接形式标明文章原始出处和作者信息及版权声明
网址: http://xumingming.sinaapp.com/756/twitter-storm-drpc/
本文翻译自: https://github.com/nathanmarz/storm/wiki/Distributed-RPC 。
Storm里面引入DRPC主要是利用storm的实时计算能力来并行化CPU intensive的计算。DRPC的storm topology以函数的参数流作为输入,而把这些函数调用的返回值作为topology的输出流。
DRPC其实不能算是storm本身的一个特性, 它是通过组合storm的原语spout,bolt, topology而成的一种模式(pattern)。本来应该把DRPC单独打成一个包的, 但是DRPC实在是太有用了,所以我们我们把它和storm捆绑在一起。
概览
Distributed RPC是由一个”DPRC Server”协调的(storm自带了一个实现)。
DRPC服务器协调
1) 接收一个RPC请求。
2) 发送请求到storm topology
3) 从storm topology接收结果。
4) 把结果发回给等待的客户端。
从客户端的角度来看一个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来跟等待的客户端匹配上,唤醒这个客户端并且把结果发送给它。
LinearDRPCTopologyBuilder
Storm自带了一个称作 LinearDRPCTopologyBuilder 的topology builder, 它把实现DRPC的几乎所有步骤都自动化了。这些步骤包括:
- 设置spout
- 把结果返回给DRPC服务器
- 给bolt提供有限聚合几组tuples的能力
让我们看一个简单的例子。下面是一个把输入参数后面添加一个”!”的DRPC topology的实现:
public static class ExclaimBolt implements IBasicBolt { public void prepare(Map conf, TopologyContext context) { } public void execute(Tuple tuple, BasicOutputCollector collector) { String input = tuple.getString(1); collector.emit(new Values(tuple.getValue(0), input + "!")); } public void cleanup() { } 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函数的名字。一个DRPC服务器可以协调很多函数,函数与函数之间靠函数名字来区分。你声明的第一个bolt会接收两维tuple,tuple的第一个field是request-id,第二个field是这个请求的参数。 LinearDRPCTopologyBuilder
同时要求我们topology的最后一个bolt发射一个二维tuple: 第一个field是request-id, 第二个field是这个函数的结果。最后所有中间tuple的第一个field必须是request-id。
在这里例子里面 ExclaimBolt
简单地在输入tuple的第二个field后面再添加一个”!”,其余的事情都由 LinearDRPCTopologyBuilder
帮我们搞定:链接到DRPC服务器,并且把结果发回。
本地模式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服务器,跟 LocalClusterLinearTopologyBuilder
有单独的方法来创建本地的topology和远程的topology。在本地模式里面LocalDRPC
对象不和任何端口绑定,所以我们的topology对象需要知道和谁交互。这就是为什么createLocalTopology
方法接受一个 LocalDRPC
对象作为输入的原因。
把topology启动了之后,你就可以通过调用 LocalDRPC
对象的 execute
来调用RPC方法了。
远程模式DRPC
在一个真实集群上面DRPC也是非常简单的,有三个步骤:
- 启动DRPC服务器
- 配置DRPC服务器的地址
- 提交DRPC topology到storm集群里面去。
我们可以通过下面的 storm
脚本命令来启动DRPC服务器:
bin/storm drpc
接着, 你需要让你的storm集群知道你的DRPC服务器在哪里。 DRPCSpout
需要这个地址从而可以从DRPC服务器来接收函数调用。这个可以配置在 storm.yaml
或者通过代码的方式配置在topology里面。通过 storm.yaml
配置是这样的:
drpc.servers:
- "drpc1.foo.com"
- "drpc2.foo.com"
最后,你通过 StormSubmitter
对象来提交DRPC topology — 跟你提交其它topology没有区别。如果要以远程的方式运行上面的例子,用下面的代码:
StormSubmitter.submitTopology( "exclamation-drpc", conf, builder.createRemoteTopology() );
我们用 createRemoteTopology
方法来创建运行在真实集群上的DRPC topology。
一个更复杂的例子
上面的DRPC例子只是为了介绍DRPC概念的一个简单的例子。下面让我们看一个复杂的、确实需要storm的并行计算能力的例子, 这个例子计算twitter上面一个url的reach值。
首先介绍一下什么是reach值,要计算一个URL的reach值,我们需要:
- 获取所有微薄里面包含这个URL的人
- 获取这些人的粉丝
- 把这些粉丝去重
- 获取这些去重之后的粉丝个数 — 这就是reach
一个简单的reach计算可能会有成千上万个数据库调用,并且可能设计到百万数量级的微薄用户。这个确实可以说是CPU intensive的计算了。你会看到的是,在storm上面来实现这个是非常非常的简单。在单台机器上面, 一个reach计算可能需要花费几分钟。而在一个storm集群里面,即时是最难的URL, 也只需要几秒。
一个reach topolgoy的例子可以在 这里 找到(storm-starter)。reach topology是这样定义的:
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"));
这个topology分四步执行:
GetTweeters
获取所发微薄里面包含制定URL的所有用户。它接收输入流:[id, url]
, 它输出:[id, tweeter]
. 每一个URL tuple会对应到很多tweeter
tuple。GetFollowers
获取这些tweeter的粉丝。它接收输入流:[id, tweeter]
, 它输出:[id, follower]
PartialUniquer
通过粉丝的id来group粉丝。这使得相同的粉丝会被引导到同一个task。因此不同的task接收到的粉丝是不同的 — 从而起到去重的作用。它的输出流:[id, count]
即输出这个task上统计的粉丝个数。- 最后,
CountAggregator
接收到所有的局部数量, 把它们加起来就算出了我们要的reach值。
我们来看一下 PartialUniquer
的实现:
public static class PartialUniquer implements IRichBolt, FinishedCallback { OutputCollector _collector; Map<Object, Set<String>> _sets = new HashMap<Object, Set<String>>(); public void prepare(Map conf, TopologyContext context, OutputCollector collector) { _collector = collector; } public void execute(Tuple tuple) { Object id = tuple.getValue(0); Set<String> curr = _sets.get(id); if(curr==null) { curr = new HashSet<String>(); _sets.put(id, curr); } curr.add(tuple.getString(1)); _collector.ack(tuple); } public void cleanup() { } public void finishedId(Object id) { Set<String> curr = _sets.remove(id); int count; if(curr!=null) { count = curr.size(); } else { count = 0; } _collector.emit(new Values(id, count)); } public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("id", "partial-count")); } }
当 PartialUniquer
在 execute
方法里面接收到一个 粉丝tuple
的时候, 它把这个tuple添加到当前request-id对应的 Set
里面去。
PartialUniquer
同时也实现了 FinishedCallback
接口, 实现这个接口是告诉 LinearDRPCTopologyBuilder
它想在接收到某个request-id的所有tuple之后得到通知,回调函数则是finishedId 方法。在这个回调函数里面 PartialUniquer
发射当前这个request-id在这个task上的粉丝数量。
在这个简单接口的背后,我们是使用 CoordinatedBolt
来检测什么时候一个bolt接收到某个request的所有的tuple的。 CoordinatedBolt
是利用direct stream来实现这种协调的。
这个topology的其余部分就非常的明了了。我们可以看到的是reach计算的每个步骤都是并行计算出来的,而且实现这个DRPC的topology是那么的简单。
非线性DRPC Topology
LinearDRPCTopologyBuilder
只能搞定"线性"的DRPC topology。所谓的线性就是说你的计算过程是一步接着一步, 串联。我们不难想象还有其它的可能 -- 并联(回想一下初中物理里面学的并联电路吧), 现在你如果想解决这种这种并联的case的话, 那么你需要自己去使用 CoordinatedBolt
来处理所有的事情了。如果真的有这种use case的话, 在mailing list上大家讨论一下吧。
LinearDRPCTopologyBuilder的工作原理
- DRPCSpout发射tuple:
[args, return-info]
。return-info
包含DRPC服务器的主机地址,端口以及当前请求的request-id - DRPC Topology包含以下元素:
- DRPCSpout
- PrepareRequest(生成request-id, return info以及args)
- CoordinatedBolt
- JoinResult -- 组合结果和return info
- ReturnResult -- 连接到DRPC服务器并且返回结果
- LinearDRPCTopologyBuilder是利用storm的原语来构建高层抽象的很好的例子。
高级特性
- 如何利用KeyedFairBolt来同时处理多个请求
- 如何直接使用CoordinatedBolt
一个更复杂的例子的全部代码
/** * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file * distributed with this work for additional information * regarding copyright ownership. The ASF licenses this file * to you under the Apache License, Version 2.0 (the * "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package cn.ljh.storm.drpc; import org.apache.storm.Config; import org.apache.storm.LocalCluster; import org.apache.storm.LocalDRPC; import org.apache.storm.StormSubmitter; import org.apache.storm.coordination.BatchOutputCollector; import org.apache.storm.drpc.LinearDRPCTopologyBuilder; import org.apache.storm.task.TopologyContext; import org.apache.storm.topology.BasicOutputCollector; import org.apache.storm.topology.OutputFieldsDeclarer; import org.apache.storm.topology.base.BaseBasicBolt; import org.apache.storm.topology.base.BaseBatchBolt; import org.apache.storm.tuple.Fields; import org.apache.storm.tuple.Tuple; import org.apache.storm.tuple.Values; import java.util.*; /** * This is a good example of doing complex Distributed RPC on top of Storm. This program creates a topology that can * compute the reach for any URL on Twitter in realtime by parallelizing the whole computation. * <p/> * Reach is the number of unique people exposed to a URL on Twitter. To compute reach, you have to get all the people * who tweeted the URL, get all the followers of all those people, unique that set of followers, and then count the * unique set. It‘s an intense computation that can involve thousands of database calls and tens of millions of follower * records. * <p/> * This Storm topology does every piece of that computation in parallel, turning what would be a computation that takes * minutes on a single machine into one that takes just a couple seconds. * <p/> * For the purposes of demonstration, this topology replaces the use of actual DBs with in-memory hashmaps. * * @see <a href="http://storm.apache.org/documentation/Distributed-RPC.html">Distributed RPC</a> */ public class ReachTopology { public static Map<String, List<String>> TWEETERS_DB = new HashMap<String, List<String>>() {{ put("foo.com/blog/1", Arrays.asList("sally", "bob", "tim", "george", "nathan")); put("engineering.twitter.com/blog/5", Arrays.asList("adam", "david", "sally", "nathan")); put("tech.backtype.com/blog/123", Arrays.asList("tim", "mike", "john")); }}; public static Map<String, List<String>> FOLLOWERS_DB = new HashMap<String, List<String>>() {{ put("sally", Arrays.asList("bob", "tim", "alice", "adam", "jim", "chris", "jai")); put("bob", Arrays.asList("sally", "nathan", "jim", "mary", "david", "vivian")); put("tim", Arrays.asList("alex")); put("nathan", Arrays.asList("sally", "bob", "adam", "harry", "chris", "vivian", "emily", "jordan")); put("adam", Arrays.asList("david", "carissa")); put("mike", Arrays.asList("john", "bob")); put("john", Arrays.asList("alice", "nathan", "jim", "mike", "bob")); }}; public static class GetTweeters extends BaseBasicBolt { public void execute(Tuple tuple, BasicOutputCollector collector) { Object id = tuple.getValue(0); String url = tuple.getString(1); List<String> tweeters = TWEETERS_DB.get(url); if (tweeters != null) { for (String tweeter : tweeters) { collector.emit(new Values(id, tweeter)); } } } public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("id", "tweeter")); } } public static class GetFollowers extends BaseBasicBolt { public void execute(Tuple tuple, BasicOutputCollector collector) { Object id = tuple.getValue(0); String tweeter = tuple.getString(1); List<String> followers = FOLLOWERS_DB.get(tweeter); if (followers != null) { for (String follower : followers) { collector.emit(new Values(id, follower)); } } } public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("id", "follower")); } } public static class PartialUniquer extends BaseBatchBolt { BatchOutputCollector _collector; Object _id; Set<String> _followers = new HashSet<String>(); public void prepare(Map conf, TopologyContext context, BatchOutputCollector collector, Object id) { _collector = collector; _id = id; } public void execute(Tuple tuple) { //利用set的特性来去重。 _followers.add(tuple.getString(1)); } public void finishBatch() { //同一个task处理完了相同id的tuple之后调用。 _collector.emit(new Values(_id, _followers.size())); } public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("id", "partial-count")); } } public static class CountAggregator extends BaseBatchBolt { BatchOutputCollector _collector; Object _id; int _count = 0; public void prepare(Map conf, TopologyContext context, BatchOutputCollector collector, Object id) { _collector = collector; _id = id; } public void execute(Tuple tuple) { _count += tuple.getInteger(1); } public void finishBatch() { _collector.emit(new Values(_id, _count)); } public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("id", "reach")); } } public static LinearDRPCTopologyBuilder construct() { LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder("reach"); builder.addBolt(new GetTweeters(), 4); builder.addBolt(new GetFollowers(), 12).shuffleGrouping(); builder.addBolt(new PartialUniquer(), 6).fieldsGrouping(new Fields("id", "follower")); builder.addBolt(new CountAggregator(), 3).fieldsGrouping(new Fields("id")); return builder; } public static void main(String[] args) throws Exception { LinearDRPCTopologyBuilder builder = construct(); Config conf = new Config(); if (args == null || args.length == 0) { conf.setMaxTaskParallelism(3); LocalDRPC drpc = new LocalDRPC(); LocalCluster cluster = new LocalCluster(); cluster.submitTopology("reach-drpc", conf, builder.createLocalTopology(drpc)); String[] urlsToTry = new String[]{ "foo.com/blog/1", "engineering.twitter.com/blog/5", "notaurl.com" }; for (String url : urlsToTry) { System.out.println("Reach of " + url + ": " + drpc.execute("reach", url)); } cluster.shutdown(); drpc.shutdown(); } else { conf.setNumWorkers(6); StormSubmitter.submitTopologyWithProgressBar(args[0], conf, builder.createRemoteTopology()); } } }
以上是关于Storm入门Twitter Storm: DRPC简介的主要内容,如果未能解决你的问题,请参考以下文章
Storm入门Twitter Storm源代码分析之CoordinatedBolt
Twitter Storm系列flume-ng+Kafka+Storm+HDFS 实时系统搭建