Learning Storm Packt 2014

相似文件 换一批

是 否 +2 论坛币

k人 参与回答

经管之家送您一份

应届毕业生专属福利!

求职就业群

赵安豆老师微信：zhaoandou666

经管之家联合CDA

送您一个全额奖学金名额~ !

立即领取

感谢您参与论坛问题回答

经管之家送您两个论坛币！

+2 论坛币

About This Book Integrate Storm with other Big Data technologies like Hadoop, HBase, and Apache Kafka Explore log processing and machine learning using Storm Step-by-step and easy-to-understand guide to effortlessly create applications with Storm Who This Book Is For If you are a Java developer who wants to enter into the world of real-time stream processing applications using Apache Storm, then this book is for you. No previous experience in Storm is required as this book starts from the basics. After finishing this book, you will be able to develop not-so-complex Storm applications. In Detail Starting with the very basics of Storm, you will learn how to set up Storm on a single machine and move on to deploying Storm on your cluster. You will understand how Kafka can be integrated with Storm using the Kafka spout. You will then proceed to explore the Trident abstraction tool with Storm to perform stateful stream processing, guaranteeing single message processing in every topology. You will move ahead to learn how to integrate Hadoop with Storm. Next, you will learn how to integrate Storm with other well-known Big Data technologies such as HBase, Redis, and Kafka to realize the full potential of Storm. Finally, you will perform in-depth case studies on Apache log processing and machine learning with a focus on Storm, and through these case studies, you will discover Storm's realm of possibilities Editorial ReviewsAbout the AuthorAnkit Jain Ankit Jain holds a Bachelor's degree in Computer Science Engineering. He has 4 years of experience in designing and architecting solutions for the Big Data domain and has been involved with several complex engagements. His technical strengths include Hadoop, Storm, S4, HBase, Hive, Sqoop, Flume, ElasticSearch, Machine Learning, Kafka, Spring, Java, and J2EE. He is currently employed with Impetus Infotech Pvt. Ltd. He also shares his thoughts on his personal blog at http://ankitasblogger.blogspot.in/. You can follow him on Twitter at @mynameisanky. He spends most of his time reading books and playing with different technologies. When not at work, he spends time with his family and friends watching movies and playing games. Product Details Paperback: 233 pages Publisher: Packt Publishing - ebooks Account (August 26, 2014) Language: English ISBN-10: 1783981326 ISBN-13: 978-1783981328 Product Dimensions: 9.2 x 7.5 x 0.5 inches Shipping Weight: 1.2 pounds (View shipping rates and policies) Average Customer Review: Be the first to review this item Amazon Best Sellers Rank: #352,081 in Books (See Top 100 in Books) 本帖隐藏的内容 Learning Storm Packt 2014.pdf (3.88 MB, 需要: 20 个论坛币) 2014-10-3 20:21:33 上传 需要: 20 个论坛币  [购买]

---

扫码加我 拉你入群

请注明：姓名-公司-职位

以便审核进群资格，未注明则拒绝

关键词：Learning earning Learn Storm ning 2014 experience developer learning previous

很好的书

1. package com.learningstorm.storm_example;
   
2. 
   
3. import backtype.storm.Config;
   
4. import backtype.storm.StormSubmitter;
   
5. import backtype.storm.generated.AlreadyAliveException;
   
6. import backtype.storm.generated.InvalidTopologyException;
   
7. import backtype.storm.topology.TopologyBuilder;
   
8. 
   
9. public class LearningStormSingleNodeTopology {
   
10.         public static void main(String[] args) {
    
11.                 TopologyBuilder builder = new TopologyBuilder();
    
12.                 builder.setSpout("LearningStormSpout", new LearningStormSpout(), 2);
    
13.                 builder.setBolt("LearningStormBolt", new LearningStormBolt(), 4).shuffleGrouping("LearningStormSpout");
    
14. 
    
15.                 Config conf = new Config();
    
16.                 conf.setNumWorkers(3);
    
17.                
    
18.                 try {
    
19.                         // args[0] is the name of submitted topology
    
20.                         StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
    
21.                 }catch(AlreadyAliveException alreadyAliveException) {
    
22.                         System.out.println(alreadyAliveException);
    
23.                 } catch (InvalidTopologyException invalidTopologyException) {
    
24.                         System.out.println(invalidTopologyException);
    
25.                 }
    
26.         }
    
27. }

复制代码

1. package com.learningstorm.storm_example;
   
2. 
   
3. import java.util.HashMap;
   
4. import java.util.Map;
   
5. import java.util.Random;
   
6. 
   
7. import backtype.storm.spout.SpoutOutputCollector;
   
8. import backtype.storm.task.TopologyContext;
   
9. import backtype.storm.topology.OutputFieldsDeclarer;
   
10. import backtype.storm.topology.base.BaseRichSpout;
    
11. import backtype.storm.tuple.Fields;
    
12. import backtype.storm.tuple.Values;
    
13. 
    
14. public class LearningStormSpout extends BaseRichSpout{
    
15.         private static final long serialVersionUID = 1L;
    
16.         private SpoutOutputCollector spoutOutputCollector;
    
17.         private static final Map<Integer, String> map = new HashMap<Integer, String>();
    
18.             static {
    
19.                 map.put(0, "google");
    
20.                 map.put(1, "facebook");
    
21.                 map.put(2, "twitter");
    
22.                 map.put(3, "youtube");
    
23.                 map.put(4, "linkedin");
    
24.     }
    
25.         public void open(Map conf, TopologyContext context,
    
26.                         SpoutOutputCollector spoutOutputCollector) {
    
27.                 // Open the spout
    
28.                 this.spoutOutputCollector = spoutOutputCollector;
    
29.         }
    
30.                
    
31.         public void nextTuple() {
    
32.                 // Storm cluster repeatedly call this method to emit the continuous
    
33.                 // stream of tuples.
    
34.                 final Random rand = new Random();
    
35.                 // generate the random number from 0 to 4.
    
36.                 int randomNumber = rand.nextInt(5);
    
37.                 spoutOutputCollector.emit(new Values(map.get(randomNumber)));
    
38. 
    
39.         }
    
40. 
    
41.         public void declareOutputFields(OutputFieldsDeclarer declarer) {
    
42.                 // emit the tuple with field name "site"
    
43.                 declarer.declare(new Fields("site"));
    
44.         }
    
45. }

复制代码

1. package com.learningstorm.storm_example;
   
2. 
   
3. import backtype.storm.Config;
   
4. import backtype.storm.LocalCluster;
   
5. import backtype.storm.generated.AlreadyAliveException;
   
6. import backtype.storm.generated.InvalidTopologyException;
   
7. import backtype.storm.topology.TopologyBuilder;
   
8. 
   
9. public class LearningStormTopology {
   
10.         public static void main(String[] args) throws AlreadyAliveException,
    
11.                         InvalidTopologyException {
    
12.                 TopologyBuilder builder = new TopologyBuilder();
    
13.                 // set the spout class
    
14.                 builder.setSpout("LearningStormSpout", new LearningStormSpout(), 2);
    
15.                 // set the bolt class
    
16.                 builder.setBolt("LearningStormBolt", new LearningStormBolt(), 4).shuffleGrouping("LearningStormSpout");
    
17. 
    
18.                 Config conf = new Config();
    
19.                 conf.setDebug(true);
    
20.                 // create an instance of LocalCluster class for
    
21.                 // executing topology in local mode.
    
22.                 LocalCluster cluster = new LocalCluster();
    
23. 
    
24.                 // LearningStormTopolgy is the name of submitted topology.
    
25.                 cluster.submitTopology("LearningStormToplogy", conf,
    
26.                                 builder.createTopology());
    
27.                 try {
    
28.                         Thread.sleep(10000);
    
29.                 } catch (Exception exception) {
    
30.                         System.out.println("Thread interrupted exception : " + exception);
    
31.                 }
    
32.                 // kill the LearningStormTopology
    
33.                 cluster.killTopology("LearningStormToplogy");
    
34.                 // shutdown the storm test cluster
    
35.                 cluster.shutdown();
    
36. 
    
37.         }
    
38. }

复制代码

1. package com.learningstorm.grouping;
   
2. 
   
3. import java.io.Serializable;
   
4. import java.util.List;
   
5. import java.util.Map;
   
6. 
   
7. import backtype.storm.generated.GlobalStreamId;
   
8. import backtype.storm.grouping.CustomStreamGrouping;
   
9. import backtype.storm.task.WorkerTopologyContext;
   
10. 
    
11. import com.google.common.collect.ImmutableList;
    
12. import com.google.common.collect.ImmutableMap;
    
13. 
    
14. public class CategoryGrouping implements CustomStreamGrouping, Serializable {
    
15.         private static final Map<String, Integer> categories = ImmutableMap.of(
    
16.                 "Financial", 0,
    
17.                 "Medical", 1,
    
18.                 "FMCG", 2,
    
19.                 "Electornics", 3
    
20.         );
    
21. 
    
22.         private int tasks = 0;
    
23. 
    
24.         public void prepare(WorkerTopologyContext context, GlobalStreamId stream,
    
25.                         List<Integer> targetTasks) {
    
26.                 tasks = targetTasks.size();
    
27.         }
    
28. 
    
29.         public List<Integer> chooseTasks(int taskId, List<Object> values) {
    
30.                 String category = (String) values.get(0);
    
31.                 return ImmutableList.of(categories.get(category) % tasks);
    
32.         }
    
33. }

复制代码

1. package com.learningstorm.monitoring;
   
2. 
   
3. import java.util.Iterator;
   
4. import java.util.Map;
   
5. import java.util.Set;
   
6. 
   
7. import backtype.storm.generated.BoltStats;
   
8. import backtype.storm.generated.ExecutorSpecificStats;
   
9. import backtype.storm.generated.ExecutorStats;
   
10. import backtype.storm.generated.ExecutorSummary;
    
11. import backtype.storm.generated.GlobalStreamId;
    
12. import backtype.storm.generated.Nimbus.Client;
    
13. import backtype.storm.generated.TopologyInfo;
    
14. 
    
15. public class BoltStatistics {
    
16. 
    
17.         private static final String DEFAULT = "default";
    
18.         private static final String ALL_TIME = ":all-time";
    
19. 
    
20.         public void printBoltStatistics(String topologyId) {
    
21. 
    
22.                 try {
    
23.                         ThriftClient thriftClient = new ThriftClient();
    
24.                         // Get the Nimbus thrift server client
    
25.                         Client client = thriftClient.getClient();
    
26.                        
    
27.                         // Get the information of given topology
    
28.                         TopologyInfo topologyInfo = client.getTopologyInfo(topologyId);
    
29.                         Iterator<ExecutorSummary> executorSummaryIterator = topologyInfo
    
30.                                         .get_executors_iterator();
    
31.                         while (executorSummaryIterator.hasNext()) {
    
32.                                 // get the executor
    
33.                                 ExecutorSummary executorSummary = executorSummaryIterator.next();
    
34.                                 ExecutorStats executorStats = executorSummary.get_stats();
    
35.                                 if (executorStats != null) {
    
36.                                         ExecutorSpecificStats executorSpecificStats = executorStats
    
37.                                                         .get_specific();
    
38.                                         String componentId = executorSummary.get_component_id();
    
39.                                         if (executorSpecificStats.is_set_bolt()) {
    
40.                                                 BoltStats boltStats = executorSpecificStats.get_bolt();
    
41.                                                 System.out.println("*************************************");
    
42.                                                 System.out.println("Component ID of Bolt " + componentId);
    
43.                                                 System.out.println("Transferred: "
    
44.                                                                 + getAllTimeStat(
    
45.                                                                                 executorStats.get_transferred(),
    
46.                                                                                 ALL_TIME));
    
47.                                                 System.out.println("Emitted: "
    
48.                                                                 + getAllTimeStat(executorStats.get_emitted(),
    
49.                                                                                 ALL_TIME));
    
50.                                                 System.out.println("Acked: "
    
51.                                                                 + getBoltStats(
    
52.                                                                                 boltStats.get_acked(), ALL_TIME));
    
53.                                                 System.out.println("Failed: "
    
54.                                                                 + getBoltStats(
    
55.                                                                                 boltStats.get_failed(), ALL_TIME));
    
56.                                                 System.out.println("Executed : "
    
57.                                                                 + getBoltStats(
    
58.                                                                                 boltStats.get_executed(), ALL_TIME));
    
59.                                                 System.out.println("*************************************");
    
60.                                         }
    
61.                                 }
    
62.                         }
    
63.                 } catch (Exception exception) {
    
64.                         throw new RuntimeException("Error occurred while fetching the bolt information :"+exception);
    
65.                 }
    
66.         }
    
67. 
    
68.         private static Long getAllTimeStat(Map<String, Map<String, Long>> map,
    
69.                         String statName) {
    
70.                 if (map != null) {
    
71.                         Long statValue = null;
    
72.                         Map<String, Long> tempMap = map.get(statName);
    
73.                         statValue = tempMap.get(DEFAULT);
    
74.                         return statValue;
    
75.                 }
    
76.                 return 0L;
    
77.         }
    
78. 
    
79.         public static Long getBoltStats(
    
80.                         Map<String, Map<GlobalStreamId, Long>> map, String statName) {
    
81.                 if (map != null) {
    
82.                         Long statValue = null;
    
83.                         Map<GlobalStreamId, Long> tempMap = map.get(statName);
    
84.                         Set<GlobalStreamId> key = tempMap.keySet();
    
85.                         if (key.size() > 0) {
    
86.                                 Iterator<GlobalStreamId> iterator = key.iterator();
    
87.                                 statValue = tempMap.get(iterator.next());
    
88.                         }
    
89.                         return statValue;
    
90.                 }
    
91.                 return 0L;
    
92.         }
    
93.        
    
94.         public static void main(String[] args) {
    
95.                 new BoltStatistics().printBoltStatistics("LearningStormClusterTopology-1-1393847956");
    
96.         }
    
97.        
    
98. }

复制代码

1. package com.learningstorm.monitoring;
   
2. 
   
3. import backtype.storm.generated.Nimbus.Client;
   
4. 
   
5. public class killTopology {
   
6.        
   
7.         public void kill(String topologyId) {
   
8.                 try {
   
9.                 ThriftClient thriftClient = new ThriftClient();
   
10.                 // Get the nimbus thrift client
    
11.                 Client client = thriftClient.getClient();
    
12.                 // kill the given topology
    
13.                 client.killTopology(topologyId);
    
14.                
    
15.                 }catch (Exception exception) {
    
16.                         throw new RuntimeException("Error occurred while killing the topology : "+exception);
    
17.                 }
    
18.         }
    
19.        
    
20.         public static void main(String[] args) {
    
21.                 new killTopology().kill("topologyId");
    
22.         }
    
23. }

复制代码

1. package com.learningstorm.monitoring;
   
2. 
   
3. import backtype.storm.generated.Nimbus.Client;
   
4. 
   
5. public class NimbusConfiguration {
   
6.        
   
7.         public void printNimbusStats() {
   
8.                 try {
   
9.                 ThriftClient thriftClient = new ThriftClient();
   
10.                 Client client = thriftClient.getClient();
    
11.                 String nimbusConiguration = client.getNimbusConf();
    
12.                 System.out.println("*************************************");
    
13.                 System.out.println("Nimbus Configuration : "+nimbusConiguration);
    
14.                 System.out.println("*************************************");
    
15.                 }catch(Exception exception) {
    
16.                         throw new RuntimeException("Error occurred while fetching the Nimbus statistics : ");
    
17.                 }
    
18.         }
    
19.        
    
20.         public static void main(String[] args) {
    
21.                 new NimbusConfiguration().printNimbusStats();
    
22.         }       
    
23. }

复制代码