Calculating Movies Ratings Distribution With Apache Flink

相似文件 换一批

是 否 +2 论坛币

k人 参与回答

经管之家送您一份

应届毕业生专属福利!

求职就业群

赵安豆老师微信：zhaoandou666

经管之家联合CDA

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

立即领取

感谢您参与论坛问题回答

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

+2 论坛币

1. If you’ve been following recent news in the Big Data world, you’ve probably heard about Apache Flink. This platform for batch and stream processing, which is built on a few significant technical innovations, can become a real game changer and it is starting to compete with existing products like Apache Spark.
   
2. 
   
3. In this post, I would like to show how to implement a simple batch processing algorithm using Apache Flink. We will work with a dataset of movie ratings and will produce a distribution of user ratings. In the process, I’ll show few tricks that you can use to improve the performance of your Flink applications.

复制代码

本帖隐藏的内容

Calculating Movies Ratings Distribution With Apache Flink.pdf (319.59 KB)

2017-5-17 10:45:00 上传

扫码加我 拉你入群

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

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

分享0 收藏0 回帖

关键词：distribution Calculating Ratings Movies RATING Movies

1. Create the Project
   
2. 
   
3. Creating an Apache Flink project is pretty straightforward. The Apache Flink developers created a project template for us, so all we need to do is to use the Maven archetype:generate command:
   
4. 
   
5. mvn archetype:generate
   
6.    -DarchetypeGroupId=org.apache.flink
   
7.    -DarchetypeArtifactId=flink-quickstart-java
   
8.    -DarchetypeCatalog=https://repository.apache.org/content/repositories/snapshots/
   
9.    -DarchetypeVersion=1.1.3
   
10. 
    
11. It will generate a pom.xml file and several example Flink applications. To write our own, we need to create a new Java class with a main method. It will work in both development mode and on a Flink cluster.

复制代码

1. DataSource<String> file = env.readTextFile("ml-latest-small/ratings.csv");
   
2. 
   
3. DataSet<Tuple2<IntValue, Integer>> ratings = file.flatMap(new ExtractRating());
   
4. 
   
5. private static class ExtractRating implements FlatMapFunction<String, Tuple2<IntValue, Integer>> {
   
6. 
   
7.   // Mutable int field to reuse to reduce GC pressure
   
8. 
   
9.   IntValue ratingValue = new IntValue();
   
10. 
    
11. 
    
12.   // Reuse rating value and result tuple
    
13. 
    
14. 
    
15.   Tuple2<IntValue, Integer> result = new Tuple2<>(ratingValue, 1);
    
16. 
    
17. 
    
18.   @Override public void flatMap(String s, Collector<Tuple2<IntValue, Integer>> collector) throws Exception {
    
19. 
    
20.     // Every line contains comma separated values // user id | item id | rating | timestamp
    
21. 
    
22.     String[] split = s.split(",");
    
23. 
    
24.     String ratingStr = split[2];
    
25. 
    
26.     // Ignore CSV header
    
27. 
    
28.     if (!ratingStr.equals("rating")) {
    
29. 
    
30.       int rating = (int) Double.parseDouble(split[2]);
    
31. 
    
32.       ratingValue.setValue(rating);
    
33. 
    
34.       collector.collect(result);
    
35. 
    
36.     }
    
37. 
    
38.   }
    
39. 
    
40. }

复制代码

1. file.flatMap(new ExtractRating()) .groupBy(0) .reduceGroup(new SumRatingCount());
   
2. 
   
3. private static class SumRatingCount implements GroupReduceFunction<Tuple2<IntValue, Integer>, Tuple2<IntValue, Integer>> {
   
4.   @Override public void reduce(Iterable<Tuple2<IntValue, Integer>> iterable, Collector<Tuple2<IntValue, Integer>> collector) throws Exception {
   
5.     IntValue rating = null;
   
6.     int ratingsCount = 0;
   
7.     for (Tuple2<IntValue, Integer> tuple : iterable) {
   
8.       rating = tuple.f0;
   
9.       ratingsCount += tuple.f1;
   
10. 
    
11.     }
    
12.     collector.collect(new Tuple2<>(rating, ratingsCount));
    
13. 
    
14.   }
    
15. 
    
16. }

复制代码

thanks..........

谢谢楼主分享！