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

说明

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

socket服务

需要启动 socket TCP 服务器

linux

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

nc -lk 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.java.functions.KeySelector;
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.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

public class TestDataStream2 {
    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);
        SingleOutputStreamOperator<Tuple2<String, Integer>> soso=dataStream.flatMap(new Splitter());
        KeyedStream<Tuple2<String, Integer>,String> ks=soso.keyBy(new keyByImpl());//流处理中的分组

        SingleOutputStreamOperator<Tuple2<String, Integer>> soso2=ks.sum(1);//按照第二个元素聚合

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

    /**
     * 拆分、并记为1，每个元素是元祖，格式： (单词,1)
     * FlatMapFunction<String, Tuple2<String, Integer>> 泛型解释：
     *      String：传入的是字符串
     *      Tuple2<String, Integer>，集合的泛型。对字符串进行处理，封装成Tuple类型，格式：(单词,1)，然后放到该集合中
     */
    public static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
        /**
         *
         * @param sentence 表示每一行数据
         * @param out  传出的集合
         */
        @Override
        public void flatMap(String sentence, Collector<Tuple2<String, Integer>> out) throws Exception {
            for (String word: sentence.split(" ")) {
                out.collect(new Tuple2<String, Integer>(word, 1));
            }
        }
    }

    /**
     根据第一个元素（即：单词）进行分组
     KeySelector<Tuple2<String, Integer>,String> 泛型解释：
        Tuple2<String, Integer>：传入的是 Tuple类型，也就是上面封装的Tuple，格式： (单词,1)
        String：传出类型，也就是Tuple中的单词，根据该单词进行分组
     */
    public static class keyByImpl implements KeySelector<Tuple2<String, Integer>,String>{
        /**
         *
         * @param t：传入的元祖，格式：(单词,1)
         * @return
         * @throws Exception
         */
        @Override
        public String getKey(Tuple2<String, Integer> t) throws Exception {
            return t.f0;
        }
    }
}

接口泛型规律：

• 第一个泛型表示传入数据的类型
• 第二个泛型表示传出数据的类型

测试

在 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_1IX1zF4hogxX.html