系统示例应用

/usr/local/flink-1.18.1/bin/flink run /usr/local/flies/streaming/SocketWindowWordCount.jar --port 9010

nc -l 9010
asd asd sdfsf sdf sdfsdagd sdf

---

单次统计示例工程

cd C:\Dev\IdeaProjectsmvn archetype:generate -DarchetypeGroupId=org.apache.flink -DarchetypeArtifactId=flink-quickstart-java -DarchetypeVersion=1.18.1

 Define value for property 'groupId':Define value for property 'artifactId':Define value for property 'version' 1.0-SNAPSHOT: :Define value for property 'package' : :com.eduflink-example1.0.0com.edu.flink

package com.edu.flink;import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;import java.time.Duration;public class WindowWordCount {public static void main(String[] args) throws Exception {//设置运行时环境StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();//设置输入流，并执行数据流的处理和转换env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);DataStream<Tuple2<String, Integer>> dataStream = env.socketTextStream("192.168.18.128", 9000).flatMap(new Splitter()).keyBy(0).timeWindow(Time.seconds(5)).sum(1);dataStream.assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(3)));//设置输出流dataStream.print();//执行程序env.execute("Window WordCount");System.out.print("finished...");}public static class Splitter implements FlatMapFunction<String, Tuple2<String,Integer>> {@Overridepublic void flatMap(String sentence, Collector<Tuple2<String, Integer>> out)throws Exception {for (String word : sentence.split(" ")) {out.collect(new Tuple2<String, Integer>(word, 1));}}}}