[Case Study]Word Count using Java

是 否 +2 论坛币

k人 参与回答

经管之家送您一份

应届毕业生专属福利!

求职就业群

赵安豆老师微信：zhaoandou666

经管之家联合CDA

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

立即领取

感谢您参与论坛问题回答

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

+2 论坛币

1. package org.apache.spark.examples.streaming;
   
2. 
   
3. import java.util.HashMap;
   
4. import java.util.HashSet;
   
5. import java.util.Arrays;
   
6. import java.util.regex.Pattern;
   
7. 
   
8. import scala.Tuple2;
   
9. 
   
10. import com.google.common.collect.Lists;
    
11. import kafka.serializer.StringDecoder;
    
12. 
    
13. import org.apache.spark.SparkConf;
    
14. import org.apache.spark.api.java.function.*;
    
15. import org.apache.spark.streaming.api.java.*;
    
16. import org.apache.spark.streaming.kafka.KafkaUtils;
    
17. import org.apache.spark.streaming.Durations;
    
18. 
    
19. /**
    
20. * Consumes messages from one or more topics in Kafka and does wordcount.
    
21. * Usage: DirectKafkaWordCount <brokers> <topics>
    
22. *   <brokers> is a list of one or more Kafka brokers
    
23. *   <topics> is a list of one or more kafka topics to consume from
    
24. *
    
25. * Example:
    
26. *    $ bin/run-example streaming.KafkaWordCount broker1-host:port,broker2-host:port topic1,topic2
    
27. */
    
28. 
    
29. public final class JavaDirectKafkaWordCount {
    
30.   private static final Pattern SPACE = Pattern.compile(" ");
    
31. 
    
32.   public static void main(String[] args) {
    
33.     if (args.length < 2) {
    
34.       System.err.println("Usage: DirectKafkaWordCount <brokers> <topics>\n" +
    
35.           "  <brokers> is a list of one or more Kafka brokers\n" +
    
36.           "  <topics> is a list of one or more kafka topics to consume from\n\n");
    
37.       System.exit(1);
    
38.     }
    
39. 
    
40.     StreamingExamples.setStreamingLogLevels();
    
41. 
    
42.     String brokers = args[0];
    
43.     String topics = args[1];
    
44. 
    
45.     // Create context with 2 second batch interval
    
46.     SparkConf sparkConf = new SparkConf().setAppName("JavaDirectKafkaWordCount");
    
47.     JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, Durations.seconds(2));
    
48. 
    
49.     HashSet<String> topicsSet = new HashSet<String>(Arrays.asList(topics.split(",")));
    
50.     HashMap<String, String> kafkaParams = new HashMap<String, String>();
    
51.     kafkaParams.put("metadata.broker.list", brokers);
    
52. 
    
53.     // Create direct kafka stream with brokers and topics
    
54.     JavaPairInputDStream<String, String> messages = KafkaUtils.createDirectStream(
    
55.         jssc,
    
56.         String.class,
    
57.         String.class,
    
58.         StringDecoder.class,
    
59.         StringDecoder.class,
    
60.         kafkaParams,
    
61.         topicsSet
    
62.     );
    
63. 
    
64.     // Get the lines, split them into words, count the words and print
    
65.     JavaDStream<String> lines = messages.map(new Function<Tuple2<String, String>, String>() {
    
66.       @Override
    
67.       public String call(Tuple2<String, String> tuple2) {
    
68.         return tuple2._2();
    
69.       }
    
70.     });
    
71.     JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
    
72.       @Override
    
73.       public Iterable<String> call(String x) {
    
74.         return Lists.newArrayList(SPACE.split(x));
    
75.       }
    
76.     });
    
77.     JavaPairDStream<String, Integer> wordCounts = words.mapToPair(
    
78.       new PairFunction<String, String, Integer>() {
    
79.         @Override
    
80.         public Tuple2<String, Integer> call(String s) {
    
81.           return new Tuple2<String, Integer>(s, 1);
    
82.         }
    
83.       }).reduceByKey(
    
84.         new Function2<Integer, Integer, Integer>() {
    
85.         @Override
    
86.         public Integer call(Integer i1, Integer i2) {
    
87.           return i1 + i2;
    
88.         }
    
89.       });
    
90.     wordCounts.print();
    
91. 
    
92.     // Start the computation
    
93.     jssc.start();
    
94.     jssc.awaitTermination();
    
95.   }
    
96. }

复制代码

扫码加我 拉你入群

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

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

分享0 收藏0 回帖

关键词：Case study Using Count study Case package import Java Word