转载请注明出处:http://blog.csdn.net/lonelytrooper/article/details/12435641

The Bolts

首先我们看一下该topology中的标准bolts:

public class UserSplitterBoltimplementsIBasicBolt{

private static final longserialVersionUID=1L;

@Override

public voiddeclareOutputFields(OutputFieldsDeclarerdeclarer) {

declarer.declareStream("users",newFields("txid","tweet_id","user"));

}

@Override

public Map<String,Object>getComponentConfiguration() {

return null;

}

@Override

publicvoidprepare(Map stormConf,TopologyContext context) {

}

@Override

publicvoidexecute(Tuple input,BasicOutputCollector collector) {

String tweet =input.getStringByField("tweet");

String tweetId =input.getStringByField("tweet_id");

StringTokenizer strTok =newStringTokenizer(tweet," ");

TransactionAttempt tx = (TransactionAttempt)input.getValueByField("txid");

HashSet<String>users= newHashSet<String>();

while(strTok.hasMoreTokens()) {

String user =strTok.nextToken();

// Ensure this is an actual user, and that it's not repeated in the tweet

if(user.startsWith("@") && !users.contains(user)) {

collector.emit("users",newValues(tx,tweetId,user));

users.add(user);

}

}

}

@Override

publicvoidcleanup() {

}

}

正如本章前边提到的,UserSplitterBolt接收元组,解析tweets的文本,并发送@后边的单词或者Twitter用户。HashtagSplitterBolt以一种非常简单的方式工作。

public class HashtagSplitterBoltimplementsIBasicBolt{

private static final longserialVersionUID=1L;

@Override

public voiddeclareOutputFields(OutputFieldsDeclarerdeclarer) {

declarer.declareStream("hashtags",newFields("txid","tweet_id","hashtag"));

}

@Override

public Map<String,Object>getComponentConfiguration() {

return null;

}

@Override

public voidprepare(Map stormConf,TopologyContext context) {

}

@Override

public voidexecute(Tuple input,BasicOutputCollector collector) {

String tweet =input.getStringByField("tweet");

String tweetId =input.getStringByField("tweet_id");

StringTokenizer strTok =newStringTokenizer(tweet," ");

TransactionAttempt tx = (TransactionAttempt)input.getValueByField("txid");

HashSet<String>words= newHashSet<String>();

while(strTok.hasMoreTokens()) {

String word =strTok.nextToken();

if(word.startsWith("#") && !words.contains(word)) {

collector.emit("hashtags",newValues(tx,tweetId,word));

words.add(word);

}

}

}

@Override

publicvoidcleanup() {

}

}

我们现在看下在UserHashtagJoinBolt中发生了什么。首先要注意到的最重要的事情是它是一个BaseBatchBolt。这意味着会对接收到的元组执行execute方法但不会发送任何新的元组。逐步的,当批次结束的时候,Storm会调用finishBatch方法。

public voidexecute(Tuple tuple) {

String source =tuple.getSourceStreamId();

String tweetId =tuple.getStringByField("tweet_id");

if("hashtags".equals(source)) {

String hashtag =tuple.getStringByField("hashtag");

add(tweetHashtags,tweetId,hashtag);

} else if("users".equals(source)) {

String user =tuple.getStringByField("user");

add(userTweets,user,tweetId);

}

}

因为你需要将一条tweet中所有的标签与该tweet中提到的用户关联起来并且计数他们出现的次数,你需要对前边bolt的两条流做连接。对整个批次都这样处理,一旦完成了,finishBatch方法会被调用。

@Override

publicvoidfinishBatch() {

for(String user:userTweets.keySet()) {

Set<String>tweets= getUserTweets(user);

HashMap<String,Integer>hashtagsCounter =new HashMap<String,Integer>();

for(String tweet:tweets) {

Set<String>hashtags= getTweetHashtags(tweet);

if(hashtags!=null) {

for(String hashtag:hashtags) {

Integer count =hashtagsCounter.get(hashtag);

if(count==null)

;

count ++;

hashtagsCounter.put(hashtag,count);

}

}

}

for (String hashtag:hashtagsCounter.keySet()) {

int count=hashtagsCounter.get(hashtag);

collector.emit(newValues(id,user,hashtag, count));

}

}

}

在该方法中,对每一个用户--标签以及它出现的次数,生成并发射一个元组。

你可以在GitHub看到完整的可下载的代码。

提交者 Bolts

正如你已经知道的,在topology中批量的元组被协调器和发射器发送。这些批量的元组被并行的处理,并没有特定的顺序。

coordinator bolts或者是实现了ICommitter接口的特殊批量bolts,或者它在TransactionalTopologyBuilder中被用setCommiterBolt方法设置过。它与常规的批量bolts的主要不同在于当该批次准备好被提交时会执行提交者 bolts的finishBatch方法。这在所有前边的事务被成功的提交后会发生。另外,finishBatch方法被顺序的执行。所以,当事务ID为1的批次和事务ID为2的批次在topology中被并行的处理时,正在处理事务ID为2的批次的提交者bolt的finishBatch方法只有在事务ID为1的批次的finishBatch方法结束并且没有任何错误的情况下才会被执行。

该类的实现如下:

public class RedisCommiterCommiterBoltextendsBaseTransactionalBolt

implements ICommitter{

public static final String LAST_COMMITED_TRANSACTION_FIELD="LAST_COMMIT";

TransactionAttempt id;

BatchOutputCollector collector;

Jedis jedis;

@Override

public voidprepare(Map conf,TopologyContext context,

BatchOutputCollector collector,TransactionAttempt id) {

this.id=id;

this.collector=collector;

this.jedis=newJedis("localhost");

}

HashMap<String,Long>hashtags = new HashMap<String,Long>();

HashMap<String,Long>users = newHashMap<String,Long>();

HashMap<String,Long>usersHashtags =new HashMap<String,Long>();

private voidcount(HashMap<String,Long>map, String key,intcount) {

Long value =map.get(key);

if(value==null)

;

value +=count;

map.put(key,value);

}

@Override

public voidexecute(Tuple tuple) {

String origin =tuple.getSourceComponent();

if("users-splitter".equals(origin)) {

String user =tuple.getStringByField("user");

);

} else if("hashtag-splitter".equals(origin)) {

String hashtag =tuple.getStringByField("hashtag");

);

} else if("user-hashtag-merger".equals(origin)) {

String hashtag =tuple.getStringByField("hashtag");

String user =tuple.getStringByField("user");

String key =user+ ":" + hashtag;

Integer count =tuple.getIntegerByField("count");

count(usersHashtags,key,count);

}

}

@Override

publicvoidfinishBatch() {

String lastCommitedTransaction = jedis.get(LAST_COMMITED_TRANSACTION_FIELD);

String currentTransaction =""+id.getTransactionId();

if(currentTransaction.equals(lastCommitedTransaction))

return;

Transaction multi =jedis.multi();

multi.set(LAST_COMMITED_TRANSACTION_FIELD,currentTransaction);

Set<String>keys= hashtags.keySet();

for(String hashtag:keys) {

Long count =hashtags.get(hashtag);

multi.hincrBy("hashtags",hashtag,count);

}

keys =users.keySet();

for(String user:keys) {

Long count = users.get(user);

multi.hincrBy("users",user,count);

}

keys =usersHashtags.keySet();

for(String key:keys) {

Long count =usersHashtags.get(key);

multi.hincrBy("users_hashtags",key,count);

}

multi.exec();

}

@Override

publicvoiddeclareOutputFields(OutputFieldsDeclarer declarer) {

}

}

这些都很直观,但是在finishBatch方法中有一个非常重要的细节。

...

multi.set(LAST_COMMITED_TRANSACTION_FIELD,currentTransaction);

...

这里你正在存储上一个被提交的事务ID到数据库。你为什么要那样做?记住当一个事务失败时,如果有必要的话Storm将重放它足够多次。如果你不确定你已经处理过该事务,那么你可以高估,这样整个topology的事务性含义都没意义了。所以记住:存储上一个被提交的事务ID并且提交前核对它。

getting start with storm 翻译 第八章 part-2的更多相关文章

  1. getting start with storm 翻译 第八章 part-1

    转载请注明出处:http://blog.csdn.net/lonelytrooper/article/details/12434915 第八章 事务性Topologies 在Storm中,正如本书前边 ...

  2. 大数据入门第十六天——流式计算之storm详解(一)入门与集群安装

    一.概述 今天起就正式进入了流式计算.这里先解释一下流式计算的概念 离线计算 离线计算:批量获取数据.批量传输数据.周期性批量计算数据.数据展示 代表技术:Sqoop批量导入数据.HDFS批量存储数据 ...

  3. The Django Book

    The Django Book Table of contents 2.0, English -> Chinese Django book 2.0 的中文翻译. 最近更新 - 贡献者 方便自己也 ...

  4. 深入理解Magento – 第四章 – 模型和ORM基础

    深入理解Magento 作者:Alan Storm 翻译:Hailong Zhang 第四章 – 模型和ORM基础 对于任何一个MVC架构,模型(Model)层的实现都是占据了很大一部分.对于Mage ...

  5. 深入理解Magento – 第三章 – 布局,块和模板

    深入理解Magento 作者:Alan Storm 翻译:Hailong Zhang 第三章 – 布局,块和模板 我们接着研究Magento.根据我们第二章讲的Magento MVC的架构,我们接下来 ...

  6. 深入理解Magento – 第二章 – Magento请求分发与控制器

    深入理解Magento 作者:Alan Storm 翻译:Hailong Zhang 第二章 – Magento请求分发与控制器 Model-View-Controller (MVC) ,模型-视图- ...

  7. 深入理解Magento - 第一章 - Magento强大的配置系统

    深入理解Magento 作者:Alan Storm翻译:zhlmmc 前言第一章 - Magento强大的配置系统第二章 - Magento请求分发与控制器第三章 - 布局,块和模板第四章 - 模型和 ...

  8. 《Entity Framework 6 Recipes》中文翻译系列 (42) ------ 第八章 POCO之使用POCO

    翻译的初衷以及为什么选择<Entity Framework 6 Recipes>来学习,请看本系列开篇 第八章 POCO 对象不应该知道如何保存它们,加载它们或者过滤它们.这是软件开发中熟 ...

  9. 《Entity Framework 6 Recipes》中文翻译系列 (46) ------ 第八章 POCO之领域对象测试和仓储测试

    翻译的初衷以及为什么选择<Entity Framework 6 Recipes>来学习,请看本系列开篇 8-8  测试领域对象 问题 你想为领域对象创建单元测试. 这主要用于,测试特定的数 ...

随机推荐

  1. 进度条ProgressDialog

    1.效果图 public void click(View view) { final ProgressDialog pdDialog = new ProgressDialog(this); //设置标 ...

  2. HTML5 WEB SQL

    var database = openDatabase("student1", "1.0", "学生表", 1024 * 1024, fun ...

  3. iOS常用库之Masonry

    简单介绍 Masonry 源码地址:https://github.com/Masonry/Masonry Masonry是一个轻量级的布局框架 拥有自己的描述语法 采用更优雅的链式语法封装自动布局 简 ...

  4. 手把手教你在VirtualBox中与主机共享文件夹

    安装VirtualBox为了共享文件夹,折腾了一晚上!网上的很多资料都不是很全面,这里就全面的总结一下,如果有其他的疑问,可以留言多多交流. VirtualBox下载地址,版本为5.1.2 设置共享文 ...

  5. mysql 插入中文时出现ERROR 1366 (HY000): Incorrect string value: &#39;\xC0\xEE\xCB\xC4&#39; for column &#39;usern ame&#39; at row 1

    1 环境: MySQL Server 6.0  命令行工具 2 问题 :  插入中文字符数据出现如下错误: ERROR 1366 (HY000): Incorrect string value: '\ ...

  6. Nagios安装部署和介绍(一)

    一.软件版本下载 Nagios版本下载地址: http://prdownloads.sourceforge.net/sourceforge/nagios/ http://sourceforge.net ...

  7. SSIS -&gt;&gt; Script Debugging and Troubleshooting

    Breakpoint是调试过程中最重要的手段,不仅对于Script Task和Script Component,对于任何其他的组件也是如此.可以在某个Event(如OnError)触发的时候设置断点来 ...

  8. POJ 3701 概率DP

    给定2^n 支足球队进行比赛,n<=7. 队伍两两之间有一个获胜的概率,求每一个队伍赢得最后比赛的概率是多少? 状态其实都是很显然的,一开始觉得这个问题很难啊,不会.dp[i][j] 表示第i支 ...

  9. css 背景图片拉伸[转]

    http://www.jeasyuicn.com/css-background-image-stretching.html background-image:url(bg.png); -moz-bac ...

  10. typedef和define

    typedef int INT; #define INTPTR1 (int*) typedef是用来声明类型别名的,在实际编写代码过程使用typedef往往是为了增加代码的可读性. #define是一 ...