./bin/start-cluster.sh 

 1 package flinkjob;
 2
 3 import org.apache.flink.api.common.functions.FlatMapFunction;
 4 import org.apache.flink.api.common.functions.ReduceFunction;
 5 import org.apache.flink.api.java.utils.ParameterTool;
 6 import org.apache.flink.streaming.api.datastream.DataStream;
 7 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 8 import org.apache.flink.streaming.api.windowing.time.Time;
 9 import org.apache.flink.util.Collector;
10
11 /**
12  * Created by adrian.wu on 2018/12/17.
13  */
14 public class SocketWindowWordCount {
15     public static class WordWithCount {
16
17         public String word;
18         public long count;
19
20         public WordWithCount() {}
21
22         public WordWithCount(String word, long count) {
23             this.word = word;
24             this.count = count;
25         }
26
27         @Override
28         public String toString() {
29             return word + " : " + count;
30         }
31     }
32
33
34     public static void main(String[] args) throws Exception{
35         final int port;
36         try {
37             //得到提交时候的参数
38             final ParameterTool params = ParameterTool.fromArgs(args);
39             //得到端口号，因为这个例子是监听9000端口的例子
40             port = params.getInt("port");
41         } catch (Exception e) {
42             System.err.println("No port specified. Please run 'SocketWindowWordCount --port <port>'");
43             return;
44         }
45         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
46
47         //数据单元 DataStream
48         DataStream<String> text = env.socketTextStream("localhost", port, "\n");
49         DataStream<WordWithCount> windowCounts = text
50                 .flatMap(new FlatMapFunction<String, WordWithCount>() {   //map
51                     @Override
52                     public void flatMap(String value, Collector<WordWithCount> out) {
53                         for (String word : value.split("\\s")) {
54                             out.collect(new WordWithCount(word, 1L));
55                         }
56                     }
57                 })
58                 .keyBy("word")
59                 .timeWindow(Time.seconds(5), Time.seconds(1))  //Window function, 5秒一个window，间隔1
60                 .reduce(new ReduceFunction<WordWithCount>() {
61                     @Override
62                     public WordWithCount reduce(WordWithCount a, WordWithCount b) {  //reduce
63                         return new WordWithCount(a.word, a.count + b.count);
64                     }
65                 });
66         windowCounts.print().setParallelism(1);
67
68         env.execute("Socket Window WordCount");
69
70     }
71 }

 ./bin/flink run examples/streaming/SocketWindowWordCount.jar --port 9000 #提交job

 nc -l 9000 #监听端口

tail -f log/flink-*-taskexecutor-*.out #查看log