【Apache Flint】Flink DataStream API Programming Guide

是 否 +2 论坛币

k人 参与回答

经管之家送您一份

应届毕业生专属福利!

求职就业群

赵安豆老师微信：zhaoandou666

经管之家联合CDA

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

立即领取

感谢您参与论坛问题回答

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

+2 论坛币

本帖隐藏的内容

https://ci.apache.org/projects/flink/flink-docs-release-1.2/dev/datastream_api.html

1. Example Program
   
2. The following program is a complete, working example of streaming window word count application, that counts the words coming from a web socket in 5 second windows. You can copy & paste the code to run it locally.
   
3. 
   
4. Java
   
5. Scala
   
6. import org.apache.flink.streaming.api.scala._
   
7. import org.apache.flink.streaming.api.windowing.time.Time
   
8. 
   
9. object WindowWordCount {
   
10.   def main(args: Array[String]) {
    
11. 
    
12.     val env = StreamExecutionEnvironment.getExecutionEnvironment
    
13.     val text = env.socketTextStream("localhost", 9999)
    
14. 
    
15.     val counts = text.flatMap { _.toLowerCase.split("\\W+") filter { _.nonEmpty } }
    
16.       .map { (_, 1) }
    
17.       .keyBy(0)
    
18.       .timeWindow(Time.seconds(5))
    
19.       .sum(1)
    
20. 
    
21.     counts.print
    
22. 
    
23.     env.execute("Window Stream WordCount")
    
24.   }
    
25. }

复制代码

扫码加我 拉你入群

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

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

分享0 收藏0 回帖

关键词：Programming datastream Program apache Stream

Thanks.

1. Running an example
   
2. In order to run a Flink example, we assume you have a running Flink instance available. The “Quickstart” and “Setup” tabs in the navigation describe various ways of starting Flink.
   
3. 
   
4. The easiest way is running the ./bin/start-local.sh script, which will start a JobManager locally.
   
5. 
   
6. Each binary release of Flink contains an examples directory with jar files for each of the examples on this page.
   
7. 
   
8. To run the WordCount example, issue the following command:
   
9. 
   
10. ./bin/flink run ./examples/batch/WordCount.jar
    
11. The other examples can be started in a similar way.
    
12. 
    
13. Note that many examples run without passing any arguments for them, by using build-in data. To run WordCount with real data, you have to pass the path to the data:
    
14. 
    
15. ./bin/flink run ./examples/batch/WordCount.jar --input /path/to/some/text/data --output /path/to/result
    
16. Note that non-local file systems require a schema prefix, such as hdfs://.

复制代码

1. Word Count
   
2. WordCount is the “Hello World” of Big Data processing systems. It computes the frequency of words in a text collection. The algorithm works in two steps: First, the texts are splits the text to individual words. Second, the words are grouped and counted.
   
3. 
   
4. Java
   
5. Scala
   
6. ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
   
7. 
   
8. DataSet<String> text = env.readTextFile("/path/to/file");
   
9. 
   
10. DataSet<Tuple2<String, Integer>> counts =
    
11.         // split up the lines in pairs (2-tuples) containing: (word,1)
    
12.         text.flatMap(new Tokenizer())
    
13.         // group by the tuple field "0" and sum up tuple field "1"
    
14.         .groupBy(0)
    
15.         .sum(1);
    
16. 
    
17. counts.writeAsCsv(outputPath, "\n", " ");
    
18. 
    
19. // User-defined functions
    
20. public static class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
    
21. 
    
22.     @Override
    
23.     public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
    
24.         // normalize and split the line
    
25.         String[] tokens = value.toLowerCase().split("\\W+");
    
26. 
    
27.         // emit the pairs
    
28.         for (String token : tokens) {
    
29.             if (token.length() > 0) {
    
30.                 out.collect(new Tuple2<String, Integer>(token, 1));
    
31.             }   
    
32.         }
    
33.     }
    
34. }
    
35. The WordCount example implements the above described algorithm with input parameters: --input <path> --output <path>. As test data, any text file will do.

复制代码

1. Page Rank
   
2. The PageRank algorithm computes the “importance” of pages in a graph defined by links, which point from one pages to another page. It is an iterative graph algorithm, which means that it repeatedly applies the same computation. In each iteration, each page distributes its current rank over all its neighbors, and compute its new rank as a taxed sum of the ranks it received from its neighbors. The PageRank algorithm was popularized by the Google search engine which uses the importance of webpages to rank the results of search queries.
   
3. 
   
4. In this simple example, PageRank is implemented with a bulk iteration and a fixed number of iterations.
   
5. 
   
6. ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
   
7. 
   
8. // read the pages and initial ranks by parsing a CSV file
   
9. DataSet<Tuple2<Long, Double>> pagesWithRanks = env.readCsvFile(pagesInputPath)
   
10.                                                    .types(Long.class, Double.class)
    
11. 
    
12. // the links are encoded as an adjacency list: (page-id, Array(neighbor-ids))
    
13. DataSet<Tuple2<Long, Long[]>> pageLinkLists = getLinksDataSet(env);
    
14. 
    
15. // set iterative data set
    
16. IterativeDataSet<Tuple2<Long, Double>> iteration = pagesWithRanks.iterate(maxIterations);
    
17. 
    
18. DataSet<Tuple2<Long, Double>> newRanks = iteration
    
19.         // join pages with outgoing edges and distribute rank
    
20.         .join(pageLinkLists).where(0).equalTo(0).flatMap(new JoinVertexWithEdgesMatch())
    
21.         // collect and sum ranks
    
22.         .groupBy(0).sum(1)
    
23.         // apply dampening factor
    
24.         .map(new Dampener(DAMPENING_FACTOR, numPages));
    
25. 
    
26. DataSet<Tuple2<Long, Double>> finalPageRanks = iteration.closeWith(
    
27.         newRanks,
    
28.         newRanks.join(iteration).where(0).equalTo(0)
    
29.         // termination condition
    
30.         .filter(new EpsilonFilter()));
    
31. 
    
32. finalPageRanks.writeAsCsv(outputPath, "\n", " ");
    
33. 
    
34. // User-defined functions
    
35. 
    
36. public static final class JoinVertexWithEdgesMatch
    
37.                     implements FlatJoinFunction<Tuple2<Long, Double>, Tuple2<Long, Long[]>,
    
38.                                             Tuple2<Long, Double>> {
    
39. 
    
40.     @Override
    
41.     public void join(<Tuple2<Long, Double> page, Tuple2<Long, Long[]> adj,
    
42.                         Collector<Tuple2<Long, Double>> out) {
    
43.         Long[] neighbors = adj.f1;
    
44.         double rank = page.f1;
    
45.         double rankToDistribute = rank / ((double) neigbors.length);
    
46. 
    
47.         for (int i = 0; i < neighbors.length; i++) {
    
48.             out.collect(new Tuple2<Long, Double>(neighbors[i], rankToDistribute));
    
49.         }
    
50.     }
    
51. }
    
52. 
    
53. public static final class Dampener implements MapFunction<Tuple2<Long,Double>, Tuple2<Long,Double>> {
    
54.     private final double dampening, randomJump;
    
55. 
    
56.     public Dampener(double dampening, double numVertices) {
    
57.         this.dampening = dampening;
    
58.         this.randomJump = (1 - dampening) / numVertices;
    
59.     }
    
60. 
    
61.     @Override
    
62.     public Tuple2<Long, Double> map(Tuple2<Long, Double> value) {
    
63.         value.f1 = (value.f1 * dampening) + randomJump;
    
64.         return value;
    
65.     }
    
66. }
    
67. 
    
68. public static final class EpsilonFilter
    
69.                 implements FilterFunction<Tuple2<Tuple2<Long, Double>, Tuple2<Long, Double>>> {
    
70. 
    
71.     @Override
    
72.     public boolean filter(Tuple2<Tuple2<Long, Double>, Tuple2<Long, Double>> value) {
    
73.         return Math.abs(value.f0.f1 - value.f1.f1) > EPSILON;
    
74.     }
    
75. }

复制代码

感谢分享

1. Connected Components
   
2. The Connected Components algorithm identifies parts of a larger graph which are connected by assigning all vertices in the same connected part the same component ID. Similar to PageRank, Connected Components is an iterative algorithm. In each step, each vertex propagates its current component ID to all its neighbors. A vertex accepts the component ID from a neighbor, if it is smaller than its own component ID.
   
3. 
   
4. This implementation uses a delta iteration: Vertices that have not changed their component ID do not participate in the next step. This yields much better performance, because the later iterations typically deal only with a few outlier vertices.
   
5. 
   
6. Java
   
7. Scala
   
8. // read vertex and edge data
   
9. DataSet<Long> vertices = getVertexDataSet(env);
   
10. DataSet<Tuple2<Long, Long>> edges = getEdgeDataSet(env).flatMap(new UndirectEdge());
    
11. 
    
12. // assign the initial component IDs (equal to the vertex ID)
    
13. DataSet<Tuple2<Long, Long>> verticesWithInitialId = vertices.map(new DuplicateValue<Long>());
    
14. 
    
15. // open a delta iteration
    
16. DeltaIteration<Tuple2<Long, Long>, Tuple2<Long, Long>> iteration =
    
17.         verticesWithInitialId.iterateDelta(verticesWithInitialId, maxIterations, 0);
    
18. 
    
19. // apply the step logic:
    
20. DataSet<Tuple2<Long, Long>> changes = iteration.getWorkset()
    
21.         // join with the edges
    
22.         .join(edges).where(0).equalTo(0).with(new NeighborWithComponentIDJoin())
    
23.         // select the minimum neighbor component ID
    
24.         .groupBy(0).aggregate(Aggregations.MIN, 1)
    
25.         // update if the component ID of the candidate is smaller
    
26.         .join(iteration.getSolutionSet()).where(0).equalTo(0)
    
27.         .flatMap(new ComponentIdFilter());
    
28. 
    
29. // close the delta iteration (delta and new workset are identical)
    
30. DataSet<Tuple2<Long, Long>> result = iteration.closeWith(changes, changes);
    
31. 
    
32. // emit result
    
33. result.writeAsCsv(outputPath, "\n", " ");
    
34. 
    
35. // User-defined functions
    
36. 
    
37. public static final class DuplicateValue<T> implements MapFunction<T, Tuple2<T, T>> {
    
38. 
    
39.     @Override
    
40.     public Tuple2<T, T> map(T vertex) {
    
41.         return new Tuple2<T, T>(vertex, vertex);
    
42.     }
    
43. }
    
44. 
    
45. public static final class UndirectEdge
    
46.                     implements FlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {
    
47.     Tuple2<Long, Long> invertedEdge = new Tuple2<Long, Long>();
    
48. 
    
49.     @Override
    
50.     public void flatMap(Tuple2<Long, Long> edge, Collector<Tuple2<Long, Long>> out) {
    
51.         invertedEdge.f0 = edge.f1;
    
52.         invertedEdge.f1 = edge.f0;
    
53.         out.collect(edge);
    
54.         out.collect(invertedEdge);
    
55.     }
    
56. }
    
57. 
    
58. public static final class NeighborWithComponentIDJoin
    
59.                 implements JoinFunction<Tuple2<Long, Long>, Tuple2<Long, Long>, Tuple2<Long, Long>> {
    
60. 
    
61.     @Override
    
62.     public Tuple2<Long, Long> join(Tuple2<Long, Long> vertexWithComponent, Tuple2<Long, Long> edge) {
    
63.         return new Tuple2<Long, Long>(edge.f1, vertexWithComponent.f1);
    
64.     }
    
65. }
    
66. 
    
67. public static final class ComponentIdFilter
    
68.                     implements FlatMapFunction<Tuple2<Tuple2<Long, Long>, Tuple2<Long, Long>>,
    
69.                                             Tuple2<Long, Long>> {
    
70. 
    
71.     @Override
    
72.     public void flatMap(Tuple2<Tuple2<Long, Long>, Tuple2<Long, Long>> value,
    
73.                         Collector<Tuple2<Long, Long>> out) {
    
74.         if (value.f0.f1 < value.f1.f1) {
    
75.             out.collect(value.f0);
    
76.         }
    
77.     }
    
78. }

复制代码

谢谢分享