Flink 实现无界流

Flink 实现无界流

1. package org.example.test;
3. import org.apache.flink.api.common.functions.FlatMapFunction;
4. import org.apache.flink.api.common.functions.MapFunction;
5. import org.apache.flink.api.java.ExecutionEnvironment;
6. import org.apache.flink.api.java.operators.AggregateOperator;
7. import org.apache.flink.api.java.operators.DataSource;
8. import org.apache.flink.api.java.operators.FlatMapOperator;
9. import org.apache.flink.api.java.operators.UnsortedGrouping;
10. import org.apache.flink.api.java.tuple.Tuple2;
11. import org.apache.flink.streaming.api.datastream.DataStream;
12. import org.apache.flink.streaming.api.datastream.DataStreamSource;
13. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
14. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
15. import org.apache.flink.util.Collector;
17. /**
18. * DataSet API使用
19. */
20. public class WordCount2 {
22.     public static void main(String[] args) throws Exception {
23.         //该类主要是用于进行批处理
24.         final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
25.         //读取文本
26.         DataSource<String> stringDataStreamSource = env.readTextFile("input/test.txt");
27.         //进行ETL处理,Tuple2 是二元数组的意思
28.         FlatMapOperator<String, Tuple2<String, Integer>> stringTuple2FlatMapOperator = stringDataStreamSource.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
29.             @Override
30.             public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
31.                 String[] words = value.split(" ");
32.                 for (String word : words) {
33.                     Tuple2<String, Integer> oneTuple2 = Tuple2.of(word, 1);
34.                     out.collect(oneTuple2);
35.                 }
36.             }
37.         });
38.         //进行分组，分组字段取下标第0个
39.         UnsortedGrouping<Tuple2<String, Integer>> tuple2UnsortedGrouping =
40.                 stringTuple2FlatMapOperator.groupBy(0);
41.         //进行sum操作
42.         AggregateOperator<Tuple2<String, Integer>> sum = tuple2UnsortedGrouping.sum(1);
43.         sum.print();
44.     }
45. }

复制代码

ExecutionEnvironment 是批处置惩罚的方式，DataSource会慢慢被淘汰

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。