[译]【Storm入门指南】第六章 真实示例

1

node webapp/app.js

12
3
4
5
6
7
8
9
10
1112
13
14
15
16
17
18
19
20
2122
23
24
25
26
27
28
29
30
3132
33
34
35
36
37
38

package storm.analytics;
...

public class TopologyStarter {

public static void main(String[] args) {
Logger.getRootLogger().removeAllAppenders();

TopologyBuilder builder = new TopologyBuilder();

builder.setSpout("read-feed", new UsersNavigationSpout(), 3);

builder.setBolt("get-categ", new GetCategoryBolt(), 3)
.shuffleGrouping("read-feed");

builder.setBolt("user-history", new UserHistoryBolt(), 5)
.fieldsGrouping("get-categ", new Fields("user"));

builder.setBolt("product-categ-counter", new ProductCategoriesCounterBolt(), 5)
.fieldsGrouping("user-history", new Fields("product"));

if(!testing)
builder.setBolt("news-notifier", new NewsNotifierBolt(), 5)
.shuffleGrouping("product-categ-counter");

Config conf = new Config();
conf.setDebug(true);

conf.put("redis-host", REDIS_HOST);
conf.put("redis-port", REDIS_PORT);
conf.put("webserver", WEBSERVER);
conf.put("download-time", DOWNLOAD_TIME);

LocalCluster cluster = new LocalCluster();
cluster.submitTopology("analytics", conf, builder.createTopology());
}
...
}

12
3
4
5
6
7
8
9
10
1112
13
14
15
16
17
18
19
20
2122
23
24
25
26

package storm.analytics;
public class UsersNavigationSpout extends BaseRichSpout {
Jedis jedis;
...
@Override
public void nextTuple() {
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 = new HashMap<String, String>();
map.put("product", product);
NavigationEntry entry = new NavigationEntry(user, type, map);
collector.emit(new Values(user, entry));
}
}

@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("user", "otherdata"));
}
}

12
3
4
5
6
7
8
9
10
1112
13
14
15
16
17
18
19
20
2122
23
24
25

package storm.analytics;

public class GetCategoryBolt extends BaseBasicBolt {
private ProductsReader reader;
...
@Override
public void execute(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(new Values(entry.getUserId(), product, categ));
} catch (Exception ex) {
System.err.println("Error processing PRODUCT tuple"+ ex);
ex.printStackTrace();
}
}
}
...
}

12
3
4
5
6
7
8
9
10
1112
13
14
15
16
17
18
19

package storm.analytics.utilities;
...
public class ProductsReader {
...
public Product readItem(String id) throws Exception{
String content= jedis.get(id);
if(content == null || ("nil".equals(content)))
return null;
Object obj=JSONValue.parse(content);
JSONObject product=(JSONObject)obj;

Product i= new Product((Long)product.get("id"),
(String)product.get("title"),
(Long)product.get("price"),
(String)product.get("category"));
return i;
}
...
}

12
3
4
5
6
7
8
9
10
1112
13
14
15
16
17
18
19
20
2122
23
24
25
26

package storm.analytics;

...
public class UserHistoryBolt extends BaseRichBolt{
@Override
public void execute(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 (String other : productsNavigated) {
String [] ot = other.split(":");
String prod2 = ot[0];
String cat2 = ot[1];
collector.emit(new Values(prod1, cat2));
collector.emit(new Values(prod2, cat1));
}
addProductToHistory(user, prodKey);
}
}
}

12
3
4
5
6
7
8
9
10
1112
13
14
15
16
17
18
19
20

public class UserHistoryBolt extends BaseRichBolt{
...
private Set<String> getUserNavigationHistory(String user) {
Set<String> userHistory = usersNavigatedItems.get(user);
if(userHistory == null) {
userHistory = jedis.smembers(buildKey(user));
if(userHistory == null)
userHistory = new HashSet<String>();
usersNavigatedItems.put(user, userHistory);
}
return userHistory;
}

private void addProductToHistory(String user, String product) {
Set<String> userHistory = getUserNavigationHistory(user);
userHistory.add(product);
jedis.sadd(buildKey(user), product);
}
...
}

12
3
4
5
6
7
8
9
10
1112
13
14
15
16
17
18
19

public class ProductCategoriesCounterBolt extends BaseRichBolt {
...
@Override
public void execute(Tuple input) {
String product = input.getString(0);
String categ = input.getString(1);
int total = count(product, categ);
collector.emit(new Values(product, categ, total));
}

...
private int count(String product, String categ) {
int count = getProductCategoryCount(categ, product);
count ++;
storeProductCategoryCount(categ, product, count);
return count;
}
...
}

12
3
4
5
6
7
8
9
10
1112
13
14
15
16
17
18
19
20
2122
23
24
25
26
27
28
29
30
3132
33

package storm.analytics;
...
public class ProductCategoriesCounterBolt extends BaseRichBolt {
// ITEM:CATEGORY -> COUNT
HashMap<String, Integer> counter = new HashMap<String, Integer>();

// ITEM:CATEGORY -> COUNT
HashMap<String, Integer> pendingToSave = new HashMap<String, Integer>();
...

public int getProductCategoryCount(String categ, String product) {
Integer count = counter.get(buildLocalKey(categ, product));
if(count == null) {
String sCount = jedis.hget(buildRedisKey(product), categ);
if(sCount == null || "nil".equals(sCount)) {
count = 0;
} else {
count = Integer.valueOf(sCount);
}
}
return count;
}

...
private void storeProductCategoryCount(String categ, String product, int count) {
String key = buildLocalKey(categ, product);
counter.put(key , count);
synchronized (pendingToSave) {
pendingToSave.put(key, count);
}
}
...
}

12
3
4
5
6
7
8
9
10
1112
13
14
15
16
17
18
19
20
2122
23
24
25
26
27

package storm.analytics;

public class ProductCategoriesCounterBolt extends BaseRichBolt {
...
private void startDownloaderThread() {
TimerTask t = new TimerTask() {
@Override
public void run() {
HashMap<String, Integer> pendings;
synchronized (pendingToSave) {
pendings = pendingToSave;
pendingToSave = new HashMap<String, Integer>();
}
for (String key : pendings.keySet()) {
String[] keys = key.split(":");
String product = keys[0];
String categ = keys[1];
Integer count = pendings.get(key);
jedis.hset(buildRedisKey(product), categ, count.toString());
}
}
};
timer = new Timer("Item categories downloader");
timer.scheduleAtFixedRate(t, downloadTime, downloadTime);
}
...
}

12
3
4
5
6
7
8
9
10
1112
13
14
15
16
17
18
19
20
2122
23

package storm.analytics;
...
public class NewsNotifierBolt extends BaseRichBolt {
...
@Override
public void execute(Tuple input) {
String product = input.getString(0);
String categ = input.getString(1);
int visits = input.getInteger(2);
String content = "{ \"product\": \""+product+"\", \"categ\":\""+categ+"\",
\"visits\":"+visits+" }";
HttpPost post = new HttpPost(webserver);
try {
post.setEntity(new StringEntity(content));
HttpResponse response = client.execute(post);
org.apache.http.util.EntityUtils.consume(response.getEntity());
} catch (Exception e) {
e.printStackTrace();
reconnect();
}
}
...
}

12
3
4

> redis-cli
redis 127.0.0.1:6379> get 15
"{\"title\":\"Kids smartphone cover\",\"category\":\"Covers\",\"price\":30,\"id\":
15}"

12
3
4
5
6
7
8
9
10
1112
13

redis 127.0.0.1:6379> llen navigation
(integer) 5
redis 127.0.0.1:6379> lrange navigation 0 4
1) "{\"user\":\"59c34159-0ecb-4ef3-a56b-99150346f8d5\",\"product\":\"1\",\"type\":
\"PRODUCT\"}"
2) "{\"user\":\"59c34159-0ecb-4ef3-a56b-99150346f8d5\",\"product\":\"1\",\"type\":
\"PRODUCT\"}"
3) "{\"user\":\"59c34159-0ecb-4ef3-a56b-99150346f8d5\",\"product\":\"2\",\"type\":
\"PRODUCT\"}"
4) "{\"user\":\"59c34159-0ecb-4ef3-a56b-99150346f8d5\",\"product\":\"3\",\"type\":
\"PRODUCT\"}"
5) "{\"user\":\"59c34159-0ecb-4ef3-a56b-99150346f8d5\",\"product\":\"5\",\"type\":
\"PRODUCT\"}"

12
3
4
5

redis 127.0.0.1:6379> smembers history:59c34159-0ecb-4ef3-a56b-99150346f8d5
1) "1:Players"
2) "5:Cameras"
3) "2:Players"
4) "3:Cameras"

12
3
4
5

redis 127.0.0.1:6379> hgetall prodcnt:2
1) "Players"
2) "1"
3) "Cameras"
4) "2"

12
3
4
5
6
7
8
9
10
1112
13
14
15
16
17
18
19
20
2122
23
24
25
26
27
28
29
30
3132
33
34
35
36
37
38
39
40
4142
43
44
45
46
47
48

public abstract class AbstractAnalyticsTest extends Assert {
def jedis
static topologyStarted = false
static sync= new Object()

private void reconnect() {
jedis = new Jedis(TopologyStarter.REDIS_HOST, TopologyStarter.REDIS_PORT)
}

@Before
public void startTopology(){
synchronized(sync){
reconnect()
if(!topologyStarted){
jedis.flushAll()
populateProducts()
TopologyStarter.testing = true
TopologyStarter.main(null)
topologyStarted = true
sleep 1000
}
}
}

...
public void populateProducts() {
def testProducts = [
[id: 0, title:"Dvd player with surround sound system",
category:"Players", price: 100],
[id: 1, title:"Full HD Bluray and DVD player",
category:"Players", price:130],
[id: 2, title:"Media player with USB 2.0 input",
category:"Players", price:70],
...
[id: 21, title:"TV Wall mount bracket 50-55 Inches",
category:"Mounts", price:80]
]

testProducts.each() { product ->
def val =
"{ \"title\": \"${product.title}\" , \"category\": \"${product.category}\"," +
" \"price\": ${product.price}, \"id\": ${product.id} }"
println val
jedis.set(product.id.toString(), val.toString())
}
}
...
}

12
3
4
5
6
7
8
9
10
11

public abstract class AbstractAnalyticsTest extendsAssert {
...
public voidnavigate(user,product) {

String nav = "{\"user\": \"${user}\",\"product\": \"${product}\", \"type\":\"PRODUCT\"}".toString()
println "Pushingnavigation: ${nav}"
jedis.lpush('navigation',nav)

}
...
}

12
3
4
5
6
7
8
9
10

public abstract class AbstractAnalyticsTest extends Assert {
...
public int getProductCategoryStats(String product, String categ) {
String count = jedis.hget("prodcnt:${product}", categ)
if(count == null || "nil".equals(count))
return 0
return Integer.valueOf(count)
}
...
}

12
3
4
5
6
7
8
9
10
1112
13
14
15
16

package functional
class StatsTest extends AbstractAnalyticsTest {
@Test
public void testNoDuplication(){
navigate("1", "0") // Players
navigate("1", "1") // Players
navigate("1", "2") // Players
navigate("1", "3") // Cameras
Thread.sleep(2000) // Give two seconds for the system to process the data.
assertEquals 1, getProductCategoryStats("0", "Cameras")
assertEquals 1, getProductCategoryStats("1", "Cameras")
assertEquals 1, getProductCategoryStats("2", "Cameras")
assertEquals 2, getProductCategoryStats("0", "Players")
assertEquals 3, getProductCategoryStats("3", "Players")
}
}