上接：flink1.12.x 创建第一个工程

说明

使用 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.functions.FlatMapFunction;
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_lambda {
    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<String> dataStream = env
                .socketTextStream("hadoop1", 9999);
        /*
        根据空格拆分
        lambda表达式在运行时，会丢失类型，所以加上 returns(Types.STRING)
         */
        SingleOutputStreamOperator<String> soso=dataStream.flatMap(
                (String value, Collector<String> out) -> Arrays.stream(value.split(" ")).forEach( out :: collect)).returns(Types.STRING);
        //将每个元素的格式转为： (单词,1)
        SingleOutputStreamOperator<Tuple2<String,Integer>> soso2=soso.map( (String value ) -> Tuple2.of( value ,1)).returns(Types.TUPLE(Types.STRING,Types.INT));
        //流处理中的分组，根据第一个元素（即：单词）进行分组
        KeyedStream<Tuple2<String,Integer>,String> ks=soso2.keyBy(value -> value.f0);
        SingleOutputStreamOperator<Tuple2<String,Integer>> soso3=ks.sum(1);//聚合

        soso3.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_1IX1zINOeopJ.html