上接：flink1.12.x DataStream实现wordcount- 连接socket（TCP）服务器-lambda写法

说明

使用 DataStream 连接 socket 服务器，接收一条数据，处理一条数据

socket服务

需要启动 socket TCP 服务器

linux

登录 hadoop1 ，通过 nc 命令启动 TCP服务器，如下：

nc -l 9999

需要先安装 nc，详见 链接

java

注意：要先启动 socket服务端，因为启动 java 要连接服务器，连不上会报错

package top.malaoshi;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;

public class TestDataStream2_lambda2 {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setRuntimeMode(RuntimeExecutionMode.BATCH);//批模式
//        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);//流模式
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//自动模式，根据数据源自行判断

        DataStream<Tuple2<String, Integer>> dataStream = env
                .socketTextStream("hadoop1", 9999)
                .flatMap(
                        (String value, Collector<String> out)
                                -> Arrays.stream(value.split(" ")).forEach( out :: collect)).returns(Types.STRING)//根据空格拆分
                .map( (String value ) -> Tuple2.of( value ,1)).returns(Types.TUPLE(Types.STRING,Types.INT))//将每个元素的格式转为： (单词,1)
                .keyBy(value -> value.f0)//流处理中的分组，根据第一个元素（即：单词）进行分组
                .sum(1);//聚合
//                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))

        dataStream.print();
        //启动并等待程序结束，可传入任务名
        env.execute("Window WordCount");
    }

}

测试

在 nc 环境下，分别输入：

hadoop flink

flink yarn

hadoop yarn

在idea控制台中，显示：

8> (hadoop,1)
7> (flink,1)
2> (yarn,1)
7> (flink,2)
2> (yarn,2)
8> (hadoop,2)

接收一条数据，处理一条数据，单词的出现次数是累加的，这就是流式处理

原文出处：/show_1IX1zIUWH7gR.html