getting start with storm 翻译 第六章 part-2

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

UsersNavigationSpout

UsersNavigationSpout负责向topology输送浏览记录。每条浏览记录是一条某用户浏览某产品页的引用。它们通过web应用被存储在Redis服务器中。我们一会儿将深入它的更多细节。

为了从Redis服务器读取记录，你需要使用https://github.com/xetorthio/jedis，一个极小极简的RedisJava客户端。


接下来的代码块只显示相关部分的代码。

package storm.analytics;
public class UsersNavigationSpoutextendsBaseRichSpout{
Jedis jedis;
...
@Override
public voidnextTuple(){
String content =jedis.rpop("navigation");
if(content==null||"nil".equals(content)){
try {Thread.sleep(300);
}catch (InterruptedException e) {}
} else{
JSONObject obj=(JSONObject)JSONValue.parse(content);
String user =obj.get("user").toString();
String product =obj.get("product").toString();
String type =obj.get("type").toString();
HashMap<String,String>map=newHashMap<String,String>();
map.put("product",product);
NavigationEntry entry =newNavigationEntry(user,type,map);
collector.emit(newValues(user,entry));
}
}
@Override
public voiddeclareOutputFields(OutputFieldsDeclarerdeclarer)
{
declarer.declare(newFields("user","otherdata"));
}
}

首先spout调用jedis.rpop("navigation")来删除并返回Redis服务器”浏览”列表中最右端的元素。如果列表已空，睡眠0.3秒以此来避免忙等循环将服务器阻塞。如果找到一条记录，则解析内容(内容是JSON)并将它映射到一个NavigationEntry对象，该对象仅仅是一个包含记录信息的POJO：

﹒浏览的用户

﹒用户浏览的页面类型

﹒页面附加信息取决于类型属性。”产品”类型页面包含一个被浏览的产品ID项。

Spout通过调用collector.emit(newValues(user,entry))发射包含该信息的元组。元组的内容是topology中下一个bolt的输入：GetCategoryBolt。

GetCategoryBolt

这是一个非常简单的bolt。它唯一的职责是反序列化由前边spout发送的元组的内容。如果消息记录是关于产品页，它会使用ProductsReader帮助类来从Redis服务器加载产品信息。然后，对于每个输入的元组，它发送一个新的包含更进一步特定产品信息的元组：

﹒用户

﹒产品

﹒产品的分类

package storm.analytics;
public class GetCategoryBoltextendsBaseBasicBolt{
private ProductsReader reader;
...
@Override
public voidexecute(Tuple input,BasicOutputCollector
collector) {
NavigationEntry entry = (NavigationEntry)input.getValue(1);
if("PRODUCT".equals(entry.getPageType())){
try {
String product = (String)entry.getOtherData().get("product");
// Call the items API to get item information
Product itm =reader.readItem(product);
if(itm==null)
return ;
String categ =itm.getCategory();
collector.emit(newValues(entry.getUserId(),product,categ));
} catch(Exception ex) {
System.err.println("Error
processing PRODUCT tuple"+ ex);
ex.printStackTrace();
}
}
}
...
}
正如前边所提到的，使用ProductReader帮助类来读取特定的产品信息。
package storm.analytics.utilities;
...
public class ProductsReader{
...
public ProductreadItem(String id)throwsException{
String content=jedis.get(id);
if(content==null||("nil".equals(content)))
return null;
Object obj=JSONValue.parse(content);
JSONObject product=(JSONObject)obj;
Product i=newProduct((Long)product.get("id"),
(String)product.get("title"),
(Long)product.get("price"),
(String)product.get("category"));
return i;
}
...
}

UserHistoryBolt

UserHistoryBolt是应用的核心。它负责保存每个用户浏览的产品的踪迹并决定哪些是应该被增加的结果对。

你将使用Redis服务器来按用户存储产品的历史记录，出于性能原因，你需要在本地维护一个存储的拷贝。你通过getUserNavigationHistory(user) 和addProductToHistory(user,prodKey)方法分别来读写，以此隐藏了数据的访问细节。

package storm.analytics;
...
public class UserHistoryBoltextendsBaseRichBolt{
@Override
public voidexecute(Tuple input)
{
String user =input.getString(0);
String prod1 =input.getString(1);
String cat1 =input.getString(2);
// Product key will have category information embedded.
String prodKey =prod1+":"+cat1;
Set<String>productsNavigated=getUserNavigationHistory(user);
// If the user previously navigated this item ->ignore it
if(!productsNavigated.contains(prodKey))
{
// Otherwise update related items
for (Stringother:productsNavigated)
{
String []ot=
other.split(":");
String prod2 =ot[0];
String cat2 =ot[1];
collector.emit(newValues(prod1,cat2));
collector.emit(newValues(prod2,cat1));
}
addProductToHistory(user,prodKey);
}
}
}

注意该bolt的期望输出是发射分类关系应该被增加的产品。

看一下源代码。Bolt保存了每个用户浏览的产品的集合。注意该集合包含的是’产品：分类’
对而不仅仅是产品。那是因为在以后的调用中你需要分类信息并且如果这些信息不需要每次
都从数据库读取的话效率会更高。这是可能的，因为产品只属于一个分类并且分类在产品的生命周期中不会改变。
在读取用户之前浏览的产品集合后(包含它们的分类)，检查是否当前产品之前已经被访问过。
如果是，本条记录被忽略。如果这是用户第一次浏览该产品，遍历用户的历史浏览记录并使
用collector.emit(new Values(prod1, cat2))方法发送一个包含当前正在被浏览的产品及历史浏览记录中所有产品的分类信息的元组，使用collector.emit(new Values(prod2, cat1))发送包含其他产品及正在被浏览的产品所属分类的另一个元组。最后，添加产品及它的分类到集合。
例如，假设用户John有如下的浏览记录:
 


以及下边的需要被处理的浏览记录：



用户还未浏览8号产品，所以你需要处理它。
所以发射的元组将是：



注意左边产品与右边分类的关系应该在同一个单元中被增加。
现在，我们探索下bolt使用的持久化。
public class UserHistoryBoltextendsBaseRichBolt{
...
private Set<String>getUserNavigationHistory(String
user) {
Set<String>userHistory=usersNavigatedItems.get(user);
if(userHistory==null)
{
userHistory =jedis.smembers(buildKey(user));
if(userHistory==null)
userHistory =newHashSet<String>();
usersNavigatedItems.put(user,userHistory);
}
return userHistory;
}
private voidaddProductToHistory(String user,String
product) {
Set<String>userHistory=getUserNavigationHistory(user);
userHistory.add(product);
jedis.sadd(buildKey(user),product);
}
...
}
getUserNavigationHistory方法返回用户已访问的产品的集合。首先，尝试使用
userNavigatedItems.get(user)方法从本地内存中读取用户的历史浏览记录，如果读取不到，通
过jedis.smembers(buildKey(user))方法从Redis服务器读取并将读取数据添加到本地内存结构
usersNavigatedItems。
用户浏览新产品时，调用addProductToHistory方法来同时更新内存结构与Redis服务器。通
过userHistory.add(peoduct)更新内存结构，通过jedis.sadd(buildKey(user),product)更新Redis
服务器。
值得注意的是既然bolt在内存中按用户保存信息，那么当你并行化bolt时，在第一级对用
户使用FieldGrouping是非常重要的，否则用户历史记录的不同拷贝将不同步。