1、socket终端输入：w,q,q,q,q,q,q,q,q,q,w,w,w,w,w,w 即为9个q和7个w
使用滚动窗口打印结果为：（w,16）
public static void test02(StreamExecutionEnvironment env) {
//数据源接入
env.socketTextStream(“192.168.1.201”,9527)
.flatMap(new FlatMapFunction<String, String>() {
@Override
//数据读进来处理后，然后返回
public void flatMap(String value, Collector out) throws Exception {
String[] words = value.split(",");
//增强的for循环
for(String word:words){
out.collect(word.toLowerCase().trim());
}
}
}).filter(new FilterFunction() {
@Override
public boolean filter(String value) throws Exception {
return StringUtils.isNotEmpty(value);
}
//每个单词赋值一个1
}).map(new MapFunction<String, Tuple2<String,Integer>>() {
@Override
public Tuple2<String, Integer> map(String value) throws Exception {
return new Tuple2<>(value,1);
}
//根据key的值统计出现的次数
}).keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
@Override
public String getKey(Tuple2<String, Integer> value) throws Exception {
return value.f0;
}
}).windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5))).
sum(1).print();
}

2.socket终端同样输入：w,q,q,q,q,q,q,q,q,q,w,w,w,w,w,w 即为9个q和7个w
注释掉//windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5))).
则打印的结果是正确的。
代码如下：
public static void test02(StreamExecutionEnvironment env) {
//数据源接入
env.socketTextStream(“192.168.1.201”,9527)
.flatMap(new FlatMapFunction<String, String>() {
@Override
//数据读进来处理后，然后返回
public void flatMap(String value, Collector out) throws Exception {
String[] words = value.split(",");
//增强的for循环
for(String word:words){
out.collect(word.toLowerCase().trim());
}
}
}).filter(new FilterFunction() {
@Override
public boolean filter(String value) throws Exception {
return StringUtils.isNotEmpty(value);
}
//每个单词赋值一个1
}).map(new MapFunction<String, Tuple2<String,Integer>>() {
@Override
public Tuple2<String, Integer> map(String value) throws Exception {
return new Tuple2<>(value,1);
}
//根据key的值统计出现的次数
}).keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
@Override
public String getKey(Tuple2<String, Integer> value) throws Exception {
return value.f0;
}
}).//windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5))).
sum(1).print();
}