目的：统计text文件中以空格分隔的单词出现频数

环境：jdk8 工具：idea2019.3

创建maven项目

引入jar包

  1.         <dependency>
  2.             <groupId>org.apache.flink</groupId>
  3.             <artifactId>flink-java</artifactId>
  4.             <version>1.10.1</version>
  5.         </dependency>
  6.         <dependency>
  7.             <groupId>org.apache.flink</groupId>
  8.             <artifactId>flink-streaming-java_2.12</artifactId>
  9.             <version>1.10.1</version>
  10.         </dependency>

代码演示

DataStream用于实时数据统计

  1.   //创建流处理执行环境
  2.   StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
  3.   env.setMaxParallelism(8);//设置并行度,默认CPU核数
  5.   //读取数据
  6.   DataStream<String> input = env.readTextFile("txt的绝对路径");
  8.   //统计数据
  9.   DataStream<Tuple2<String,Integer>> res=input.flatMap(new MyFlatMapper()).keyBy(0).sum(1);
  11.   //输出打印
  12.   res.print();
  14.   //执行
  15.   env.execute();
  17.   public static class MyFlatMapper implements FlatMapFunction<String, Tuple2<String, Integer>> {
  19.         public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
  20.             String[] words = value.split(" ");
  21.             for (String word : words) {
  22.                 out.collect(new Tuple2<>(word, 1));
  23.             }
  24.         }
  25.     }

流处理单词统计.rar(4 KB)

结果

>前代表线程

  1. SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
  2. SLF4J: Defaulting to no-operation (NOP) logger implementation
  3. SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
  4. 2> (hello,1)
  5. 2> (hello,2)
  6. 4> (cc,1)
  7. 3> (word,1)
  8. 3> (java,1)
  9. 2> (hello,3)
  10. 1> (php,1)