1. 依赖

    <dependencies>
        <!--    对于java版本的flink 使用   -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
            <!--            <scope>provided</scope>-->
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.12</artifactId>
            <version>${flink.version}</version>
            <!--            <scope>provided</scope>-->
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>


        <!--   针对于scala语言使用的     -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-scala_2.12</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-scala_2.12</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>


    </dependencies>

2. 代码

• 接收来自socket（nc -l 9001）的数据
• 每两秒进行一次wc聚合

  //env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //source from socket
        DataStreamSource<String> text = env.socketTextStream("localhost", 9001);

        //trans：
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordCount =
                text.flatMap(new FlatMapFunction<String, String>() {
                    public void flatMap(String line, Collector<String> out) throws Exception {
                        String[] words = line.split(" ");
                        for (String word : words) {
                            out.collect(word);
                        }
                    }
                }).map(new MapFunction<String, Tuple2<String, Integer>>() {
                    public Tuple2<String, Integer> map(String word) throws Exception {
                        return new Tuple2<String, Integer>(word, 1);
                    }
                }).keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
                    public String getKey(Tuple2<String, Integer> tup) throws Exception {
                        return tup.f0;
                    }
                }).timeWindow(Time.seconds(2)  //两秒的窗口
                ).sum(1);

        //使用一个线程执行打印操作
        wordCount.print().setParallelism(1);

        //执行程序
        env.execute("SocketWindowWordCountJava");
 
flink datastream程序有几个组成：

1. env：创建flink job的运行环境
2. source：从哪里获取到数据
3. trans：对数据具体做哪些处理
4. sink：数据写到什么地方
5. execute：触发任务的运行