一文带你全方位(架构，原理及代码实现)了解Flink(3.2W字建议收藏)_flink 内存管理 源码

Spark Streaming

Flink

DStream

DataStream

Trasnformation

Trasnformation

Action

Sink

Task

SubTask

Pipeline

Oprator chains

DAG

DataFlow Graph

Master + Driver

JobManager

Worker + Executor

TaskManager

框架

优点

缺点

Storm

低延迟

吞吐量低、不能保证exactly-once、编程API不丰富

Spark Streaming

吞吐量高、可以保证exactly-once、编程API丰富

延迟较高

Flink

低延迟、吞吐量高、可以保证exactly-once、编程API丰富

快速迭代中,API变化比较快

#指定jobmanager的地址

jobmanager.rpc.address: 192.168.xx.xx

#指定taskmanager的可用槽位的数量

taskmanager.numberOfTaskSlots: 6

192.168.xx.xx

bin/start-cluster.sh

jps

bin/flink run -m 192.168.xx.xx:8081 -p 4 -c com.wedoctor.flink.WordCountDemo /home/pgxl/liuzc/flink-project-scala-1.0.jar --hostname 192.168.xx.xx --port 8888

heool word

#   export HADOOP_CLASSPATH=`hadoop classpath`

#   export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop

The program finished with the following exception:

java.lang.IllegalStateException: No Executor found. Please make sure to export the HADOOP_CLASSPATH environment variable or have hadoop in your classpath. For more information refer to the "Deployment & Operations" section of the official Apache Flink documentation.

at org.apache.flink.yarn.cli.FallbackYarnSessionCli.isActive(FallbackYarnSessionCli.java:59)

at org.apache.flink.client.cli.CliFrontend.validateAndGetActiveCommandLine(CliFrontend.java:1090)

at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:218)

at org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:916)

at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:992)

at org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:30)

at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:992)

./bin/flink run -m yarn-cluster -yqu root.wedw -p 4 -yjm 4096m -ytm 4096m  examples/batch/WordCount.jar

./bin/flink run -m yarn-cluster -yqu root.wedw -p 4 -yjm 4096m -ytm 4096m  /home/pgxl/liuzc/flink-project-scala-1.0.jar

杀进程

yarn application -kill application_id

查看日志

yarn  logs  -applicationId  application_id

bin/yarn-session.sh  -s 2 -jm 1024 -tm 1024 -qu root.wedw

Usage:

   Optional

     -D <arg>                        Dynamic properties

     -d,--detached                   Start detached

     -jm,--jobManagerMemory <arg>    Memory for JobManager Container with optional unit (default: MB)

     -nm,--name                      Set a custom name for the application on YARN

     -at,--applicationType           Set a custom application type on YARN

     -q,--query                      Display available YARN resources (memory, cores)

     -qu,--queue <arg>               Specify YARN queue.

     -s,--slots <arg>                Number of slots per TaskManager

     -tm,--taskManagerMemory <arg>   Memory per TaskManager Container with optional unit (default: MB)

     -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths for HA mode

bin/yarn-session.sh  -s 2 -d -jm 1024 -tm 1024 -qu root.wedw

为了正常停止Flink群集，请使用以下命令：echo "stop" | ./bin/yarn-session.sh -id <appId>。

也可以通过YARN的网络界面或实用程序杀死Flink yarn application -kill <appId>。但是请注意，杀死Flink可能不会清除所有作业工件和临时文件。

    <!-- web ui的依赖 -->

        <dependency>

            <groupId>org.apache.flink</groupId>

            <artifactId>flink-runtime-web_${scala.binary.version}</artifactId>

            <version>${flink.version}</version>

        </dependency>

StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

mvn archetype:generate \

 -DarchetypeGroupId=org.apache.flink \

 -DarchetypeArtifactId=flink-quickstart-java \

 -DarchetypeVersion=1.11.1 \

 -DgroupId=com.wedoctor.flink \

 -DartifactId=flink-project-java \

 -Dversion=1.0 \

 -Dpackage=com.wedoctor.flink \

-DinteractiveMode=false

curl https://flink.apache.org/q/quickstart.sh | bash -s 1.11.1

mvn archetype:generate \

 -DarchetypeGroupId=org.apache.flink \

 -DarchetypeArtifactId=flink-quickstart-scala \

 -DarchetypeVersion=1.11.1 \

 -DgroupId=com.wedoctor.flink \

 -DartifactId=flink-project-scala \

 -Dversion=1.0 \

 -Dpackage=com.wedoctor.flink \

-DinteractiveMode=false

curl https://flink.apache.org/q/quickstart-scala.sh | bash -s 1.11.1

package com.wedoctor.flink

import org.apache.flink.streaming.api.scala._

object WordCountDemo {

  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val lines: DataStream[String] = env.socketTextStream("192.168.xx.xx",9999)

    val words: DataStream[String] = lines.flatMap(_.split(" "))

    val wordWithOne: DataStream[(String, Int)] = words.map((_,1))

    val keyedData: KeyedStream[(String, Int), String] = wordWithOne.keyBy(_._1)

    val sumData: DataStream[(String, Int)] = keyedData.sum(1)

    sumData.print()

    env.execute("Flink WordCount")

  }

}

package com.wedoctor.flink

import org.apache.flink.api.scala._

object WordCountDemo2 {

  def main(args: Array[String]): Unit = {

    val env = ExecutionEnvironment.getExecutionEnvironment

    val words: DataSet[Int] = env.fromElements(1,2,3)

    val tt: DataSet[Int] = words.map(t=>t*2)

    tt.print()

  }

}

package com.wedoctor.flink;

import org.apache.flink.api.common.functions.MapFunction;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class MapTest {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> lines = env.socketTextStream("192.168.xx.xx", 9999);

        SingleOutputStreamOperator<String> words = lines.map(new MapFunction<String, String>() {

            @Override

            public String map(String value) throws Exception {

                return value.toUpperCase();

            }

        });

        words.print();

        env.execute();

    }

}

package com.wedoctor.flink;

import org.apache.flink.api.common.functions.FilterFunction;

import org.apache.flink.api.common.functions.RichMapFunction;

import org.apache.flink.configuration.Configuration;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class RichMapTest {

    //RichMapFunction

    //1.可以获取运行时上下文，可以得到很多的信息，subTaskIndex、状态数据等

    //2.还可以使用两个生命周期方法、open和close

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> lines = env.socketTextStream("192.168.XX.XX", 9999);

        SingleOutputStreamOperator<String> map = lines.map(new RichMapFunction<String, String>() {

            //构造对象完成后，map方法执行之前，执行一次

            @Override

            public void open(Configuration parameters) throws Exception {

                super.open(parameters);

                //此处可以建立连接

            }

            @Override

            public String map(String value) throws Exception {

                //处理数据

                return value + "222222222";

            }

            //subtask在停止之前，执行一次

            @Override

            public void close() throws Exception {

                super.close();

                //关闭连接

            }

        });

        map.print();

        env.execute();

    }

}

package com.wedoctor.flink;

import org.apache.flink.api.common.functions.FlatMapFunction;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.util.Collector;

public class FlatMapTest {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> lines = env.socketTextStream("192.168.xx.xx", 9999);

        SingleOutputStreamOperator<String> flatMap = lines.flatMap(new FlatMapFunction<String, String>() {

            @Override

            public void flatMap(String value, Collector<String> collector) throws Exception {

                String[] words = value.split(" ");

                for (String word : words) {

                    collector.collect(word);

                }

            }

        });

        flatMap.print();

        env.execute();

    }

}

package com.wedoctor.flink;

import org.apache.flink.api.common.functions.FilterFunction;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class RichMapTest {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> lines = env.socketTextStream("192.168.xx.xx", 9999);

        SingleOutputStreamOperator<String> filter = lines.filter(new FilterFunction<String>() {

            @Override

            public boolean filter(String value) throws Exception {

                return value.length() == 2;

            }

        });

        filter.print();

        env.execute();

    }

}

package com.wedoctor.flink;

import org.apache.flink.api.common.functions.FlatMapFunction;

import org.apache.flink.api.java.tuple.Tuple;

import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

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;

public class KeyByDemo {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> lines = env.socketTextStream("192.168.xx.xx", 9999);

        SingleOutputStreamOperator<Tuple2<String, Integer>> flatMap = lines.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {

            @Override

            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {

                String[] words = s.split(" ");

                for (String word : words) {

                    collector.collect(Tuple2.of(word, 1));

                }

            }

        });

        //按照单个字段分组 keyby

        KeyedStream<Tuple2<String, Integer>, Tuple> keyBy = flatMap.keyBy(0);

        KeyedStream<Tuple2<String, Integer>, String> keyBy1 = flatMap.keyBy(t -> t.f0);

        keyBy.print();

        keyBy1.print();

        env.execute();

    }

}

package com.wedoctor.flink;

import org.apache.flink.api.common.functions.MapFunction;

import org.apache.flink.api.java.tuple.Tuple;

import org.apache.flink.api.java.tuple.Tuple3;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.KeyedStream;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class KeyByDemo {

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

       // jack 01 1232

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> lines = env.socketTextStream("192.168.xx.xx", 9999);

        SingleOutputStreamOperator<Tuple3<String, String, Integer>> map = lines.map(new MapFunction<String, Tuple3<String, String, Integer>>() {

            @Override

            public Tuple3<String, String, Integer> map(String s) throws Exception {

                String[] words = s.split(" ");

                String userId = words[0];

                String monthId = words[1];

                Integer orderCnt = Integer.parseInt(words[2]);

                return Tuple3.of(userId, monthId, orderCnt);

            }

        });

        KeyedStream<Tuple3<String, String, Integer>, Tuple> key = map.keyBy(0, 1);

        SingleOutputStreamOperator<Tuple3<String, String, Integer>> summed = key.sum(2);

        summed.print();

        env.execute();

    }

}

package com.wedoctor.flink;

import org.apache.flink.api.common.functions.MapFunction;

import org.apache.flink.api.java.functions.KeySelector;

import org.apache.flink.api.java.tuple.Tuple;

import org.apache.flink.api.java.tuple.Tuple3;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.KeyedStream;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class KeyByDemo {

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

       // jack 01 1232

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> lines = env.socketTextStream("192.168.xx.xx", 9999);

        SingleOutputStreamOperator<Tuple3<String, String, Integer>> map = lines.map(new MapFunction<String, Tuple3<String, String, Integer>>() {

            @Override

            public Tuple3<String, String, Integer> map(String s) throws Exception {

                String[] words = s.split(" ");

                String userId = words[0];

                String monthId = words[1];

                Integer orderCnt = Integer.parseInt(words[2]);

                return Tuple3.of(userId, monthId, orderCnt);

            }

        });

        KeyedStream<Tuple3<String, String, Integer>, String> keyBy = map.keyBy(t -> t.f0 + t.f1);

        SingleOutputStreamOperator<Tuple3<String, String, Integer>> summed = keyBy.sum(2);

        summed.print();

        env.execute();

    }

}

package com.wedoctor.flink;

public class WordCount {

    public String word;

    public Integer count;

    public WordCount(String word, Integer count) {

        this.word = word;

        this.count = count;

    }

    public WordCount() {

    }

    public static WordCount of(String word,Integer count){

        return new WordCount(word,count);

    }

    @Override

    public String toString() {

        return "WordCount{" +

                "word='" + word + '\'' +

                ", count=" + count +

                '}';

    }

}

package com.wedoctor.flink;

public class WordCount {

    public String word;

    public Integer count;

    public WordCount(String word, Integer count) {

        this.word = word;

        this.count = count;

    }

    public WordCount() {

    }

    public static WordCount of(String word,Integer count){

        return new WordCount(word,count);

    }

    @Override

    public String toString() {

        return "WordCount{" +

                "word='" + word + '\'' +

                ", count=" + count +

                '}';

    }

}

package com.wedoctor.flink;

import org.apache.flink.api.common.functions.FlatMapFunction;

import org.apache.flink.api.common.functions.ReduceFunction;

import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.util.Collector;

public class ReduceDemo {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> lines = env.socketTextStream("192.168.xx.xx", 9999);

        SingleOutputStreamOperator<Tuple2<String, Integer>> flatMap = lines.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {

            @Override

            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {

                String[] words = s.split(" ");

                for (String word : words) {

                    collector.collect(Tuple2.of(word, 1));

                }

            }

        });

        SingleOutputStreamOperator<Tuple2<String, Integer>> reduce = flatMap.keyBy(t -> t.f0).reduce(new ReduceFunction<Tuple2<String, Integer>>() {

            @Override

            public Tuple2<String, Integer> reduce(Tuple2<String, Integer> t1, Tuple2<String, Integer> t2) throws Exception {

                return Tuple2.of(t1.f0, t1.f1 + t2.f1);

            }

        });

        reduce.print();

        env.execute();

    }

}

package com.wedoctor.flink;

import org.apache.flink.api.common.functions.MapFunction;

import org.apache.flink.api.java.functions.KeySelector;

import org.apache.flink.api.java.tuple.Tuple;

import org.apache.flink.api.java.tuple.Tuple3;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.KeyedStream;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class KeyByDemo {

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

       // jack 01 1232

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> lines = env.socketTextStream("192.168.xx.xx", 9999);

        SingleOutputStreamOperator<Tuple3<String, String, Integer>> map = lines.map(new MapFunction<String, Tuple3<String, String, Integer>>() {

            @Override

            public Tuple3<String, String, Integer> map(String s) throws Exception {

                String[] words = s.split(" ");

                String userId = words[0];

                String monthId = words[1];

                Integer orderCnt = Integer.parseInt(words[2]);

                return Tuple3.of(userId, monthId, orderCnt);

            }

        });

        KeyedStream<Tuple3<String, String, Integer>, String> keyBy = map.keyBy(t -> t.f0 + t.f1);

        SingleOutputStreamOperator<Tuple3<String, String, Integer>> summed = keyBy.sum(2);

        summed.print();

        env.execute();

    }

}

package com.wedoctor.flink;

import org.apache.flink.api.common.functions.MapFunction;

import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.KeyedStream;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class AggDemo {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> lines = env.socketTextStream("192.168.x.xx", 9999);

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndCnt = lines.map(new MapFunction<String, Tuple2<String, Integer>>() {

            @Override

            public Tuple2<String, Integer> map(String s) throws Exception {

                String[] fileds = s.split(" ");

                String word = fileds[0];

                int cnt = Integer.parseInt(fileds[1]);

                return Tuple2.of(word, cnt);

            }

        });

        //按照单个字段分组 keyby

        KeyedStream<Tuple2<String, Integer>, String> keyBy1 = wordAndCnt.keyBy(t -> t.f0);

        keyBy1.min(1).print();

        env.execute();

    }

}

package com.wedoctor.flink;

import org.apache.flink.api.common.functions.MapFunction;

import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.KeyedStream;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class AggDemo {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> lines = env.socketTextStream("192.168.xx.xx", 9999);

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndCnt = lines.map(new MapFunction<String, Tuple2<String, Integer>>() {

            @Override

            public Tuple2<String, Integer> map(String s) throws Exception {

                String[] fileds = s.split(" ");

                String word = fileds[0];

                int cnt = Integer.parseInt(fileds[1]);

                return Tuple2.of(word, cnt);

            }

        });

        //按照单个字段分组 keyby

        KeyedStream<Tuple2<String, Integer>, String> keyBy1 = wordAndCnt.keyBy(t -> t.f0);

        keyBy1.max(1).print();

        env.execute();

    }

}

package com.wedoctor.flink;

import org.apache.flink.api.common.functions.MapFunction;

import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.api.java.tuple.Tuple3;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.KeyedStream;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class AggDemo {

    //lucy 2020-05 15

    //jack 2020-02 25

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> lines = env.socketTextStream("192.168.x.xx", 9999);

        SingleOutputStreamOperator<Tuple3<String,String, Integer>> map = lines.map(new MapFunction<String, Tuple3<String,String, Integer>>() {

            @Override

            public Tuple3<String, String,Integer> map(String s) throws Exception {

                String[] fileds = s.split(" ");

                String userId = fileds[0];

                String monthId = fileds[1];

                int orderCnt = Integer.parseInt(fileds[2]);

                return Tuple3.of(userId,monthId,orderCnt);

            }

        });

        KeyedStream<Tuple3<String, String, Integer>, String> keyBy = map.keyBy(t -> t.f0);

        keyBy.minBy(2,false).print();

        env.execute();

    }

}

package com.wedoctor.flink;

import org.apache.flink.api.common.functions.MapFunction;

import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.api.java.tuple.Tuple3;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.KeyedStream;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class AggDemo {

    //lucy 2020-05 15

    //jack 2020-02 25

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> lines = env.socketTextStream("192.168.xx.xx", 9999);

        SingleOutputStreamOperator<Tuple3<String,String, Integer>> map = lines.map(new MapFunction<String, Tuple3<String,String, Integer>>() {

            @Override

            public Tuple3<String, String,Integer> map(String s) throws Exception {

                String[] fileds = s.split(" ");

                String userId = fileds[0];

                String monthId = fileds[1];

                int orderCnt = Integer.parseInt(fileds[2]);

                return Tuple3.of(userId,monthId,orderCnt);

            }

        });

        KeyedStream<Tuple3<String, String, Integer>, String> keyBy = map.keyBy(t -> t.f0);

        keyBy.maxBy(2,false).print();

        env.execute();

    }

}

package com.wedoctor.flink;

import org.apache.flink.streaming.api.datastream.DataStream;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class UnionDemo {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //调用Source创建DataStream

        DataStreamSource<Integer> s1 = env.fromElements(1, 2, 3, 4, 5);

        DataStreamSource<Integer> s2 = env.fromElements(5, 7, 8, 9, 10);

        DataStream<Integer> unioned = s1.union(s2);

        unioned.print();

        env.execute();

    }

}

package com.wedoctor.flink;

import org.apache.flink.streaming.api.datastream.AllWindowedStream;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.windowing.time.Time;

import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

public class TumblingTimeWindow {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> lines = env.socketTextStream("192.168.xx.xx", 9999);

        //默认的CountWindow是⼀个滚动窗⼝，只需要指定窗⼝⼤小即可，当元素数量达到窗口⼤小时，就会触发窗⼝的执⾏。

        SingleOutputStreamOperator<Integer> num = lines.map(Integer::parseInt);

        //划分窗口

        AllWindowedStream<Integer, TimeWindow> timeWindowAll = num.timeWindowAll(Time.seconds(5));

        //对窗口数据进行计算

        SingleOutputStreamOperator<Integer> sum = timeWindowAll.sum(0);

        sum.print();

        env.execute();

    }

}

package com.wedoctor.flink;

import org.apache.flink.api.common.typeinfo.Types;

import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.KeyedStream;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.datastream.WindowedStream;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.windowing.time.Time;

import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;

import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

public class TumblingTimeWindow2 {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> lines = env.socketTextStream("192.168.xx.xx", 9999);

        //默认的CountWindow是⼀个滚动窗⼝，只需要指定窗⼝⼤小即可，当元素数量达到窗口⼤小时，就会触发窗⼝的执⾏。

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndCount = lines.map(line -> {

            String[] fileds = line.split(",");

            return Tuple2.of(fileds[0], Integer.parseInt(fileds[1]));

        }).returns(Types.TUPLE(Types.STRING,Types.INT));

        KeyedStream<Tuple2<String, Integer>, String> keyedStream = wordAndCount.keyBy(t -> t.f0);

        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> timeWindow = keyedStream.timeWindow(Time.seconds(5));

        timeWindow.sum(1).print();

        env.execute();

    }

}

package com.wedoctor.flink;

import org.apache.flink.api.common.functions.MapFunction;

import org.apache.flink.streaming.api.TimeCharacteristic;

import org.apache.flink.streaming.api.datastream.AllWindowedStream;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;

import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;

import org.apache.flink.streaming.api.windowing.time.Time;

import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

import java.text.ParseException;

import java.text.SimpleDateFormat;

public class EventTimeTumbingWindwAllDemo {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //设置EventTime作为时间标准

        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //创建一个DataStream

        //2020-11-08 18:22:43,1

        DataStreamSource<String> lines = env.socketTextStream("192.168.xx.xx", 9999);

        //提取数据中的时间

        SingleOutputStreamOperator<String> watermarksDataStream = lines.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(0)) {

            private SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

            @Override

            public long extractTimestamp(String element) {

                long timestamp = 0;

                try {

                    timestamp = sdf.parse(element.split(",")[0]).getTime();

                } catch (ParseException e) {

                    timestamp = System.currentTimeMillis();

                }

                return timestamp;

            }

        });

        SingleOutputStreamOperator<Integer> nums = watermarksDataStream.map(new MapFunction<String, Integer>() {

            @Override

            public Integer map(String value) throws Exception {

                return Integer.parseInt(value.split(",")[1]);

            }

        });

        AllWindowedStream<Integer, TimeWindow> windowed = nums.windowAll(TumblingEventTimeWindows.of(Time.seconds(5)));

        SingleOutputStreamOperator<Integer> summed = windowed.sum(0);

        summed.print();

        env.execute();

    }

}

package com.wedoctor.flink;

import org.apache.flink.api.common.functions.MapFunction;

import org.apache.flink.api.common.typeinfo.Types;

import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.configuration.Configuration;

import org.apache.flink.streaming.api.TimeCharacteristic;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.KeyedStream;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.datastream.WindowedStream;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;

import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;

import org.apache.flink.streaming.api.windowing.time.Time;

import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

public class EventTimeTumblingWindowDemo {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        //设置EventTime作为时间标准

        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //1000,hadoop,1

        DataStreamSource<String> lines = env.socketTextStream("192.168.xx.xx", 9999);

        SingleOutputStreamOperator<String> watermarksDataStream = lines.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(0)) {

            @Override

            public long extractTimestamp(String element) {

                return Long.parseLong(element.split(",")[0]);

            }

        });

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndCount = watermarksDataStream.map(new MapFunction<String, Tuple2<String, Integer>>() {

            @Override

            public Tuple2<String, Integer> map(String value) throws Exception {

                String[] fileds = value.split(",");

                String word = fileds[1];

                int count = Integer.parseInt(fileds[2]);

                return Tuple2.of(word, count);

            }

        });

        //先分组

        KeyedStream<Tuple2<String, Integer>, String> keyed = wordAndCount.keyBy(t -> t.f0);

        //划分窗口

          WindowedStream<Tuple2<String, Integer>, String, TimeWindow> window = keyed.window(TumblingEventTimeWindows.of(Time.seconds(5)));

        SingleOutputStreamOperator<Tuple2<String, Integer>> summed = window.sum(1);

        summed.print();

        env.execute();

    }

}

package com.wedoctor.flink;

import org.apache.flink.streaming.api.datastream.AllWindowedStream;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.windowing.time.Time;

import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

public class TumblingTimeWindow {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> lines = env.socketTextStream("192.168.xx.xx", 9999);

        //默认的CountWindow是⼀个滚动窗⼝，只需要指定窗⼝⼤小即可，当元素数量达到窗口⼤小时，就会触发窗⼝的执⾏。

        SingleOutputStreamOperator<Integer> num = lines.map(Integer::parseInt);

        //划分滑动窗口

        AllWindowedStream<Integer, TimeWindow> timeWindowAll = num.timeWindowAll(Time.seconds(10),Time.seconds(5));

        //对窗口数据进行计算

        SingleOutputStreamOperator<Integer> sum = timeWindowAll.sum(0);

        sum.print();

        env.execute();

    }

}

package com.wedoctor.flink;

import org.apache.flink.api.common.typeinfo.Types;

import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.KeyedStream;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.datastream.WindowedStream;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.windowing.time.Time;

import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;

import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

public class TumblingTimeWindow2 {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> lines = env.socketTextStream("192.168.xx.xx", 9999);

        //默认的CountWindow是⼀个滚动窗⼝，只需要指定窗⼝⼤小即可，当元素数量达到窗口⼤小时，就会触发窗⼝的执⾏。

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndCount = lines.map(line -> {

            String[] fileds = line.split(",");

            return Tuple2.of(fileds[0], Integer.parseInt(fileds[1]));

        }).returns(Types.TUPLE(Types.STRING,Types.INT));

        KeyedStream<Tuple2<String, Integer>, String> keyedStream = wordAndCount.keyBy(t -> t.f0);

        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> timeWindow = keyedStream.timeWindow(Time.seconds(10),Time.seconds(5));

        timeWindow.sum(1).print();

        env.execute();

    }

}

package com.wedoctor.flink;

import org.apache.flink.configuration.Configuration;

import org.apache.flink.streaming.api.datastream.AllWindowedStream;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;

import org.apache.flink.streaming.api.windowing.time.Time;

import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

public class ProcessingTimeSessionWindowAllDemo {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        DataStreamSource<String> lines = env.socketTextStream("192.168.xx.xx", 9999);

        SingleOutputStreamOperator<Integer> nums = lines.map(Integer::parseInt);

        //不分组，划分会话窗口

        AllWindowedStream<Integer, TimeWindow> windowed = nums.windowAll(ProcessingTimeSessionWindows.withGap(Time.seconds(5)));

        //划分完窗口要调用WindowFunction对窗口内的数据进行计算

        SingleOutputStreamOperator<Integer> summed = windowed.sum(0);

        summed.print();

        env.execute();

    }

}

package com.wedoctor.flink;

import org.apache.flink.api.common.typeinfo.Types;

import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.configuration.Configuration;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.KeyedStream;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.datastream.WindowedStream;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;

import org.apache.flink.streaming.api.windowing.time.Time;

import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

public class ProcessingTimeSessionWindwDemo {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        //spark,3

        //hadoop,2

        //flink,1

        DataStreamSource<String> lines = env.socketTextStream("192.168.xx.xx", 9999);

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndCount = lines.map(line -> {

            String[] fields = line.split(",");

            return Tuple2.of(fields[0], Integer.parseInt(fields[1]));

        }).returns(Types.TUPLE(Types.STRING, Types.INT));

        //先分组

        KeyedStream<Tuple2<String, Integer>, String> keyed = wordAndCount.keyBy(t -> t.f0);

        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowed = keyed.window(ProcessingTimeSessionWindows.withGap(Time.seconds(5)));

        SingleOutputStreamOperator<Tuple2<String, Integer>> summed = windowed.sum(1);

        summed.print();

        env.execute();

    }

}

package com.wedoctor.flink;

import org.apache.flink.api.common.functions.MapFunction;

import org.apache.flink.configuration.Configuration;

import org.apache.flink.streaming.api.TimeCharacteristic;

import org.apache.flink.streaming.api.datastream.AllWindowedStream;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;

import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;

import org.apache.flink.streaming.api.windowing.time.Time;

import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

public class EventTimeSessionWindowAllDemo {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //1000,1

        DataStreamSource<String> lines = env.socketTextStream("192.168.xx.xx", 9999);

        //提取数据中的时间

        SingleOutputStreamOperator<String> watermarksDataStream = lines.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(0)) {

            @Override

            public long extractTimestamp(String element) {

                return Long.parseLong(element.split(",")[0]);

            }

        });

        SingleOutputStreamOperator<Integer> nums = watermarksDataStream.map(new MapFunction<String, Integer>() {

            @Override

            public Integer map(String value) throws Exception {

                return Integer.parseInt(value.split(",")[1]);

            }

        });

        //不分组划分窗口

        AllWindowedStream<Integer, TimeWindow> windowed = nums.windowAll(EventTimeSessionWindows.withGap(Time.seconds(5)));

        windowed.sum(0).print();

        env.execute();

    }

}

package com.wedoctor.flink;

import org.apache.flink.api.common.functions.MapFunction;

import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.configuration.Configuration;

import org.apache.flink.streaming.api.TimeCharacteristic;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.KeyedStream;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.datastream.WindowedStream;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;

import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;

import org.apache.flink.streaming.api.windowing.time.Time;

import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

public class EventTimeSessionWindowDemo {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        //设置EventTime作为时间标准

        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //1000,spark,1

        DataStreamSource<String> lines = env.socketTextStream("192.168.xx.xx", 9999);

        SingleOutputStreamOperator<String> watermarksDataStream = lines.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(0)) {

            @Override

            public long extractTimestamp(String element) {

                return Long.parseLong(element.split(",")[0]);

            }

        });

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndCount = watermarksDataStream.map(new MapFunction<String, Tuple2<String, Integer>>() {

            @Override

            public Tuple2<String, Integer> map(String value) throws Exception {

                String[] fileds = value.split(",");

                String word = fileds[1];

                int count = Integer.parseInt(fileds[2]);

                return Tuple2.of(word, count);

            }

        });

        //先分组

        KeyedStream<Tuple2<String, Integer>, String> keyed = wordAndCount.keyBy(t -> t.f0);

        //划分窗口

        //keyed.timeWindow(Time.seconds(5));

        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> window = keyed.window(EventTimeSessionWindows.withGap(Time.seconds(5)));

        SingleOutputStreamOperator<Tuple2<String, Integer>> summed = window.sum(1);

        summed.print();

        env.execute();

    }

}

package com.wedoctor.flink;

import org.apache.flink.streaming.api.datastream.AllWindowedStream;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;

public class CountWindow {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> lines = env.socketTextStream("192.168.xx.xx", 9999);

        //默认的CountWindow是⼀个滚动窗⼝，只需要指定窗⼝⼤小即可，当元素数量达到窗口⼤小时，就会触发窗⼝的执⾏。

        SingleOutputStreamOperator<Integer> num = lines.map(Integer::parseInt);

        //划分窗口

        AllWindowedStream<Integer, GlobalWindow> windowd = num.countWindowAll(5);

        //对窗口数据进行计算

        SingleOutputStreamOperator<Integer> sum = windowd.sum(0);

        sum.print();

        env.execute();

    }

}

package com.wedoctor.flink;

import org.apache.flink.api.common.typeinfo.Types;

import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.KeyedStream;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.datastream.WindowedStream;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;

public class CountWindow2 {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> lines = env.socketTextStream("192.168.xx.xx", 9999);

        //默认的CountWindow是⼀个滚动窗⼝，只需要指定窗⼝⼤小即可，当元素数量达到窗口⼤小时，就会触发窗⼝的执⾏。

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndCount = lines.map(line -> {

            String[] fileds = line.split(",");

            return Tuple2.of(fileds[0], Integer.parseInt(fileds[1]));

        }).returns(Types.TUPLE(Types.STRING,Types.INT));

        KeyedStream<Tuple2<String, Integer>, String> keyedStream = wordAndCount.keyBy(t -> t.f0);

        WindowedStream<Tuple2<String, Integer>, String, GlobalWindow> countWindow = keyedStream.countWindow(5);

        countWindow.sum(1).print();

        env.execute();

    }

}

val env = StreamExecutionEnvironment.getExecutionEnvironment

// 从调⽤时刻开始给env创建的每⼀个stream追加时间特征

env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

val stream = env.readTextFile("eventTest.txt").assignTimestampsAndWatermarks(

 new BoundedOutOfOrdernessTimestampExtractor[String](Time.milliseconds(200)) {

 override def extractTimestamp(t: String): Long = {

 // EventTime是⽇志⽣成时间，我们从⽇志中解析EventTime

 t.split(" ")(0).toLong

 }

})

[00:00:00,00:00:03)

[00:00:03,00:00:06)

...

[00:00:57,00:01:00)

[00:00:00,00:00:10)

[00:00:10,00:00:20)

...

[00:00:50,00:01:00)

package com.wedoctor.flink;

import org.apache.flink.api.common.functions.MapFunction;

import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.configuration.Configuration;

import org.apache.flink.streaming.api.TimeCharacteristic;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.KeyedStream;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.datastream.WindowedStream;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;

import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;

import org.apache.flink.streaming.api.windowing.time.Time;

import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

public class WaterMarkDemo1 {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        //设置EventTime作为时间标准

        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //1000,spark,1

        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);

        //直接对Source提取EventTime

        SingleOutputStreamOperator<String> watermarksDataStream = lines.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(2)) {

            @Override

            public long extractTimestamp(String element) {

                //当前分区中数据中最大的EventTime - 延迟时间 = 该分区的WaterMark

                return Long.parseLong(element.split(",")[0]);

            }

        });

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndCount = watermarksDataStream.map(new MapFunction<String, Tuple2<String, Integer>>() {

            @Override

            public Tuple2<String, Integer> map(String value) throws Exception {

                String[] fileds = value.split(",");

                String word = fileds[1];

                int count = Integer.parseInt(fileds[2]);

                return Tuple2.of(word, count);

            }

        });

        //先分组

        KeyedStream<Tuple2<String, Integer>, String> keyed = wordAndCount.keyBy(t -> t.f0);

        //划分窗口

        //keyed.timeWindow(Time.seconds(5));

        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> window = keyed.window(TumblingEventTimeWindows.of(Time.seconds(5)));

        SingleOutputStreamOperator<Tuple2<String, Integer>> summed = window.sum(1);

        summed.print();

        env.execute();

    }

}

// 获取执⾏环境

val env = StreamExecutionEnvironment.getExecutionEnvironment

env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

// 创建SocketSource

val stream = env.socketTextStream("localhost", 11111)

// 对stream进⾏处理并按key聚合

val streamKeyBy = stream.assignTimestampsAndWatermarks(

 new BoundedOutOfOrdernessTimestampExtractor[String](Time.milliseconds(0)) {

 override def extractTimestamp(element: String): Long = {

 val sysTime = element.split(" ")(0).toLong

 println(sysTime)

 sysTime

 }}).map(item => (item.split(" ")(1), 1)).keyBy(0)

// 引⼊滚动窗⼝

val streamWindow = streamKeyBy.window(SlidingEventTimeWindows.of(Time.seconds(10),

Time.seconds(5)))

// 执⾏聚合操作

val streamReduce = streamWindow.reduce(

 (a,b) => (a._1, a._2 + b._2)

)

// 将聚合数据写⼊⽂件

streamReduce.print

// 执⾏程序

env.execute("TumblingWindow")

package com.wedoctor.flink;

import org.apache.flink.api.common.functions.MapFunction;

import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.configuration.Configuration;

import org.apache.flink.streaming.api.TimeCharacteristic;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.KeyedStream;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.datastream.WindowedStream;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;

import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;

import org.apache.flink.streaming.api.windowing.time.Time;

import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

public class WaterMarkDemo3 {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        //设置EventTime作为时间标准

        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //1000,spark,1

        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);

        //直接对Source提取EventTime

        SingleOutputStreamOperator<String> watermarksDataStream = lines.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(2)) {

            @Override

            public long extractTimestamp(String element) {

                //当前分区中数据中最大的EventTime - 延迟时间 = 该分区的WaterMark

                return Long.parseLong(element.split(",")[0]);

            }

        });

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndCount = watermarksDataStream.map(new MapFunction<String, Tuple2<String, Integer>>() {

            @Override

            public Tuple2<String, Integer> map(String value) throws Exception {

                String[] fileds = value.split(",");

                String word = fileds[1];

                int count = Integer.parseInt(fileds[2]);

                return Tuple2.of(word, count);

            }

        });

        //先分组

        KeyedStream<Tuple2<String, Integer>, String> keyed = wordAndCount.keyBy(t -> t.f0);

        //分组后划分EventTime的SessionWindow

        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowed = keyed.window(EventTimeSessionWindows.withGap(Time.seconds(5)));

        SingleOutputStreamOperator<Tuple2<String, Integer>> summed = windowed.sum(1);

        summed.print();

        env.execute();

    }

}

Begin a new chain, starting with this operator. The two mappers will be chained, and filter will not be chained to the first mapper.

someStream.filter(...).map(...).startNewChain().map(...);

Do not chain the map operator

someStream.map(...).disableChaining();

env.setRestartStrategy(RestartStrategies.fixedDelayRestart(10, 30000));

StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();

env.enableCheckpointing(10000);

env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

//2 个 Checkpoint 之间最少是要等 500ms，也就是刚做完一个 Checkpoint。比如某个 Checkpoint 做了700ms，按照原则过 300ms 应该是做下一个 Checkpoint，因为设置了 1000ms 做一次 Checkpoint 的，但是中间的等待时间比较短，不足 500ms 了，需要多等 200ms，因此以这样的方式防止 Checkpoint 太过于频繁而导致业务处理的速度下降。

env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);

env.getCheckpointConfig().setCheckpointTimeout(6000);

//程序异常退出或人为cancel掉，不删除checkpoint数据(默认是会删除)   env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

//设置存储位置覆盖默认方式

env.setStateBackend(new FsStateBackend(args[0]));

 //广播数据的状态描述器

        MapStateDescriptor<String, String> mapStateDescriptor = new MapStateDescriptor<String, String>(

                "broadcasr-state",

                String.class,

                String.class

        );

package com.wedoctor.flink;

import com.alibaba.druid.pool.DruidDataSourceFactory;

import javax.sql.DataSource;

import java.sql.Connection;

import java.sql.SQLException;

import java.util.Properties;

public class DruidConnectionPool {

    private transient static DataSource dataSource = null;

    private transient static Properties props = new Properties();

    static {

        props.put("driverClassName", "com.mysql.jdbc.Driver");

        props.put("url", "jdbc:mysql://localhost:3306/bigdata?characterEncoding=UTF-8");

        props.put("username", "root");

        props.put("password", "123456");

        try {

            dataSource = DruidDataSourceFactory.createDataSource(props);

        } catch (Exception e) {

            e.printStackTrace();

        }

    }

    private DruidConnectionPool() {

    }

    public static Connection getConnection() throws SQLException {

        return dataSource.getConnection();

    }

}

package com.wedoctor.flink;

import org.apache.flink.api.common.ExecutionConfig;

import org.apache.flink.api.common.typeutils.base.VoidSerializer;

import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer;

import org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction;

import java.sql.Connection;

import java.sql.PreparedStatement;

import java.sql.SQLException;

public class MySqlTwoPhaseCommitSink extends TwoPhaseCommitSinkFunction<Tuple2<String, Integer>, MySqlTwoPhaseCommitSink.ConnectionState, Void> {

    public MySqlTwoPhaseCommitSink() {

        super(new KryoSerializer<>(MySqlTwoPhaseCommitSink.ConnectionState.class, new ExecutionConfig()), VoidSerializer.INSTANCE);

    }

    @Override

    protected MySqlTwoPhaseCommitSink.ConnectionState beginTransaction() throws Exception {

        System.out.println("=====> beginTransaction... ");

        Connection connection = DruidConnectionPool.getConnection();

        connection.setAutoCommit(false);

        return new ConnectionState(connection);

    }

    @Override

    protected void invoke(MySqlTwoPhaseCommitSink.ConnectionState connectionState, Tuple2<String, Integer> value, Context context) throws Exception {

        Connection connection = connectionState.connection;

        PreparedStatement pstm = connection.prepareStatement("INSERT INTO t_wordcount (word, counts) VALUES (?, ?) ON DUPLICATE KEY UPDATE counts = ?");

        pstm.setString(1, value.f0);

        pstm.setInt(2, value.f1);

        pstm.setInt(3, value.f1);

        pstm.executeUpdate();

        pstm.close();

    }

    @Override

    protected void preCommit(MySqlTwoPhaseCommitSink.ConnectionState connectionState) throws Exception {

        System.out.println("=====> preCommit... " + connectionState);

    }

    @Override

    protected void commit(MySqlTwoPhaseCommitSink.ConnectionState connectionState) {

        System.out.println("=====> commit... ");

        Connection connection = connectionState.connection;

        try {

            connection.commit();

            connection.close();

        } catch (SQLException e) {

            throw new RuntimeException("提交事物异常");

        }

    }

    @Override

    protected void abort(MySqlTwoPhaseCommitSink.ConnectionState connectionState) {

        System.out.println("=====> abort... ");

        Connection connection = connectionState.connection;

        try {

            connection.rollback();

            connection.close();

        } catch (SQLException e) {

            throw new RuntimeException("回滚事物异常");

        }

    }

    static class ConnectionState {

        private final transient Connection connection;

        ConnectionState(Connection connection) {

            this.connection = connection;

        }

    }

}

<dependency>

<groupId>org.apache.flink</groupId>

<artifactId>flink-connector-kafka_2.11</artifactId>

<version>1.10.0</version>

</dependency>

<dependency>

<groupId>org.apache.bahir</groupId>

<artifactId>flink-connector-redis_2.11</artifactId>

<version>1.1-SNAPSHOT</version>

</dependency>

package com.wedoctor.flink;

import org.apache.flink.api.common.functions.FlatMapFunction;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;

import org.apache.flink.api.common.serialization.SimpleStringSchema;

import org.apache.flink.api.java.tuple.Tuple;

import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.runtime.state.filesystem.FsStateBackend;

import org.apache.flink.streaming.api.CheckpointingMode;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.KeyedStream;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.environment.CheckpointConfig;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import org.apache.flink.streaming.connectors.redis.RedisSink;

import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;

import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;

import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;

import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

import org.apache.flink.util.Collector;

import java.util.Properties;

public class KafkaSourceToRedis {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //本地环境读取hdfs需要设置，集群上不需要

        System.setProperty("HADOOP_USER_NAME","root");

        //默认情况下，检查点被禁用。要启用检查点

        env.enableCheckpointing(30000);

        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        //程序异常退出或人为cancel掉，不删除checkpoint数据(默认是会删除)

        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        env.setStateBackend(new FsStateBackend(args[0]));

        //设置重启策略 默认不停重启

        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(10, 30000));

        Properties properties = new Properties();

        properties.setProperty("bootstrap.servers", "com.wedoctor:9092,com.wedoctor:9092,com.wedoctor:9092");

        properties.setProperty("group.id", args[1]);

        properties.setProperty("auto.offset.reset", "earliest");

        //properties.setProperty("enable.auto.commit", "false");

        //如果没有开启checkpoint功能，为了不重复读取数据，FlinkKafkaConsumer会将偏移量保存到了Kafka特殊的topic中（__consumer_offsets）

        //这种方式没法实现Exactly-Once

        FlinkKafkaConsumer<String> flinkKafkaConsumer = new FlinkKafkaConsumer<String>(args[2], new SimpleStringSchema(), properties);

        //在Checkpoint的时候将Kafka的偏移量保存到Kafka特殊的Topic中，默认是true

        flinkKafkaConsumer.setCommitOffsetsOnCheckpoints(false);

        DataStreamSource<String> lines = env.addSource(flinkKafkaConsumer);

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = lines.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {

            @Override

            public void flatMap(String line, Collector<Tuple2<String, Integer>> out) throws Exception {

                String[] words = line.split(" ");

                for (String word : words) {

                    out.collect(Tuple2.of(word, 1));

                }

            }

        });

        KeyedStream<Tuple2<String, Integer>, Tuple> keyed = wordAndOne.keyBy(0);

        SingleOutputStreamOperator<Tuple2<String, Integer>> summed = keyed.sum(1);

        //Transformation 结束

        //调用RedisSink将计算好的结果保存到Redis中

        //创建Jedis连接的配置信息

        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder()

                .setHost(args[3])

                .setPassword(args[4])

                .build();

        summed.addSink(new RedisSink<>(conf, new RedisWordCountMapper()));

        env.execute("KafkaSourceDemo");

    }

    public static class RedisWordCountMapper implements RedisMapper<Tuple2<String, Integer>> {

        @Override

        public RedisCommandDescription getCommandDescription() {

            //指定写入Redis中的方法和最外面的大key的名称

            return new RedisCommandDescription(RedisCommand.HSET, "wc");

        }

        @Override

        public String getKeyFromData(Tuple2<String, Integer> data) {

            return data.f0; //将数据中的哪个字段作为key写入

        }

        @Override

        public String getValueFromData(Tuple2<String, Integer> data) {

            return data.f1.toString(); //将数据中的哪个字段作为value写入

        }

    }

}

package com.wedoctor.flink;

import org.apache.flink.api.java.tuple.Tuple3;

import org.apache.flink.configuration.Configuration;

import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import redis.clients.jedis.Jedis;

public class MyRedisSink extends RichSinkFunction<Tuple3<String,String,String>> {

    private transient Jedis jedis;

    @Override

    public void open(Configuration parameters) throws Exception {

        super.open(parameters);

        jedis = new Jedis("192.168.1.1", 6379, 5000);

        jedis.auth("123456");

        jedis.select(0);

    }

    @Override

    public void invoke(Tuple3<String, String, String> value, Context context) throws Exception {

        if (!jedis.isConnected()){

            jedis.connect();

        }

        jedis.hset(value.f0,value.f1,value.f2);

    }

    @Override

    public void close() throws Exception {

        super.close();

        jedis.close();

    }

}

        <!-- mysql连接依赖 -->

        <dependency>

            <groupId>mysql</groupId>

            <artifactId>mysql-connector-java</artifactId>

            <version>5.1.47</version>

        </dependency>

package com.wedoctor.flink;

import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.configuration.Configuration;

import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;

import java.sql.DriverManager;

import java.sql.PreparedStatement;

public class MySqlSink extends RichSinkFunction<Tuple2<String, Integer>> {

    private Connection connection = null;

    @Override

    public void open(Configuration parameters) throws Exception {

        //可以创建数据库连接

        connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/bigdata?characterEncoding=UTF-8", "root", "123456");

    }

    @Override

    public void invoke(Tuple2<String, Integer> value, Context context) throws Exception {

        PreparedStatement preparedStatement = connection.prepareStatement("INSERT INTO test VALUES (?, ?) ON DUPLICATE KEY UPDATE counts = ?");

        preparedStatement.setString(1, value.f0);

        preparedStatement.setLong(2, value.f1);

        preparedStatement.setLong(3, value.f1);

        preparedStatement.executeUpdate();

        preparedStatement.close();

    }

    @Override

    public void close() throws Exception {

        connection.close();

    }

}

package com.wedoctor.flink;

import org.apache.flink.api.common.functions.FlatMapFunction;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;

import org.apache.flink.api.common.serialization.SimpleStringSchema;

import org.apache.flink.api.java.tuple.Tuple;

import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.runtime.state.filesystem.FsStateBackend;

import org.apache.flink.streaming.api.CheckpointingMode;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.KeyedStream;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.environment.CheckpointConfig;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import org.apache.flink.util.Collector;

import java.util.Properties;

public class KafkaSourceToMySQL {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //如果开启Checkpoint，偏移量会存储到哪呢？

        env.enableCheckpointing(30000);

        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE);

        //就是将job cancel后，依然保存对应的checkpoint数据

        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        env.setStateBackend(new FsStateBackend(args[0]));

        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(10, 30000));

        Properties properties = new Properties();

        properties.setProperty("bootstrap.servers", "node-1.51doit.cn:9092,node-2.51doit.cn:9092,node-3.51doit.cn:9092");

        properties.setProperty("group.id", args[1]);

        properties.setProperty("auto.offset.reset", "earliest");

        //properties.setProperty("enable.auto.commit", "false");

        //如果没有开启checkpoint功能，为了不重复读取数据，FlinkKafkaConsumer会将偏移量保存到了Kafka特殊的topic中（__consumer_offsets）

        //这种方式没法实现Exactly-Once

        FlinkKafkaConsumer<String> flinkKafkaConsumer = new FlinkKafkaConsumer<String>(args[2], new SimpleStringSchema(), properties);

        //在Checkpoint的时候将Kafka的偏移量保存到Kafka特殊的Topic中，默认是true

        flinkKafkaConsumer.setCommitOffsetsOnCheckpoints(false);

        DataStreamSource<String> lines = env.addSource(flinkKafkaConsumer);

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = lines.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {

            @Override

            public void flatMap(String line, Collector<Tuple2<String, Integer>> out) throws Exception {

                String[] words = line.split(" ");

                for (String word : words) {

                    out.collect(Tuple2.of(word, 1));

                }

            }

        });

        KeyedStream<Tuple2<String, Integer>, Tuple> keyed = wordAndOne.keyBy(0);

        SingleOutputStreamOperator<Tuple2<String, Integer>> summed = keyed.sum(1);

        //Transformation 结束

        //调用MySQLSink将计算好的结果保存到MySQL中

        summed.addSink(new MySqlSink());

        env.execute("KafkaSourceToMySQL");

    }

}

<!-- table api依赖 -->

<dependency>

<groupId>org.apache.flink</groupId>

<artifactId>flink-table-api-java-bridge_2.11</artifactId>

<version>1.11.2</version>

</dependency>

<dependency>

<groupId>org.apache.flink</groupId>

<artifactId>flink-table-planner_2.11</artifactId>

<version>1.11.2</version>

<!--<scope>provided</scope>-->

</dependency>

package com.wedoctor.flink;

public class WC {

    public  String word;

    public Integer cnt;

    public WC(String word, Integer cnt) {

        this.word = word;

        this.cnt = cnt;

    }

    public WC() {

    }

    public String getWord() {

        return word;

    }

    public void setWord(String word) {

        this.word = word;

    }

    public Integer getCnt() {

        return cnt;

    }

    public void setCnt(Integer cnt) {

        this.cnt = cnt;

    }

    @Override

    public String toString() {

        return "WC{" +

                "word='" + word + '\'' +

                ", cnt=" + cnt +

                '}';

    }

    public static WC of(String word, Integer cnt) {

        WC wc = new WC();

        wc.word = word;

        wc.cnt = cnt;

        return wc;

    }

}

package com.wedoctor.flink;

import org.apache.flink.api.java.DataSet;

import org.apache.flink.api.java.ExecutionEnvironment;

import org.apache.flink.table.api.Table;

import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;

public class WordCountTableApi {

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

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        BatchTableEnvironment tEnv = BatchTableEnvironment.create(env);

        DataSet<WC> input = env.fromElements(

                WC.of("spark", 1),

                WC.of("kafka", 1),

                WC.of("java", 1),

                WC.of("flink", 1),

                WC.of("flink", 1)

        );

        tEnv.registerDataSet("WordCount", input, "word, cnt");

        //执行SQL，并结果集做为一个新表

        Table table = tEnv.sqlQuery("SELECT word, SUM(cnt) as cnt FROM WordCount GROUP BY word");

       DataSet<WC> result = tEnv.toDataSet(table, WC.class);

        result.print();

    }

}

import com.alibaba.fastjson.JSON;

import com.alibaba.fastjson.JSONArray;

import com.alibaba.fastjson.JSONObject;

import org.apache.flink.api.common.functions.CoGroupFunction;

import org.apache.flink.api.common.functions.RichMapFunction;

import org.apache.flink.api.common.serialization.SimpleStringSchema;

import org.apache.flink.api.java.functions.KeySelector;

import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.api.java.utils.ParameterTool;

import org.apache.flink.configuration.Configuration;

import org.apache.flink.streaming.api.TimeCharacteristic;

import org.apache.flink.streaming.api.datastream.DataStream;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.functions.ProcessFunction;

import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;

import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;

import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;

import org.apache.flink.streaming.api.windowing.time.Time;

import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

import org.apache.flink.util.Collector;

import org.apache.flink.util.OutputTag;

import java.sql.*;

public class OrderJoinAdv {

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

        ParameterTool parameters = ParameterTool.fromPropertiesFile(args[0]);

        FlinkUtilsV2.getEnv().setParallelism(1);

        //使用EventTime作为时间标准

        FlinkUtilsV2.getEnv().setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStream<String> orderMainLinesDataStream = FlinkUtilsV2.createKafkaDataStream(parameters, "ordermain", "g1", SimpleStringSchema.class);

        DataStream<String> orderDetailLinesDataStream = FlinkUtilsV2.createKafkaDataStream(parameters, "orderdetail", "g1", SimpleStringSchema.class);

        //对数据进行解析

        SingleOutputStreamOperator<OrderMain> orderMainDataStream = orderMainLinesDataStream.process(new ProcessFunction<String, OrderMain>() {

            @Override

            public void processElement(String line, Context ctx, Collector<OrderMain> out) throws Exception {

                //flatMap+filter

                try {

                    JSONObject jsonObject = JSON.parseObject(line);

                    String type = jsonObject.getString("type");

                    if (type.equals("INSERT") || type.equals("UPDATE")) {

                        JSONArray jsonArray = jsonObject.getJSONArray("data");

                        for (int i = 0; i < jsonArray.size(); i++) {

                            OrderMain orderMain = jsonArray.getObject(i, OrderMain.class);

                            orderMain.setType(type); //设置操作类型

                            out.collect(orderMain);

                        }

                    }

                } catch (Exception e) {

                    //e.printStackTrace();

                    //记录错误的数据

                }

            }

        });

        //对数据进行解析

        SingleOutputStreamOperator<OrderDetail> orderDetailDataStream = orderDetailLinesDataStream.process(new ProcessFunction<String, OrderDetail>() {

            @Override

            public void processElement(String line, Context ctx, Collector<OrderDetail> out) throws Exception {

                //flatMap+filter

                try {

                    JSONObject jsonObject = JSON.parseObject(line);

                    String type = jsonObject.getString("type");

                    if (type.equals("INSERT") || type.equals("UPDATE")) {

                        JSONArray jsonArray = jsonObject.getJSONArray("data");

                        for (int i = 0; i < jsonArray.size(); i++) {

                            OrderDetail orderDetail = jsonArray.getObject(i, OrderDetail.class);

                            orderDetail.setType(type); //设置操作类型

                            out.collect(orderDetail);

                        }

                    }

                } catch (Exception e) {

                    //e.printStackTrace();

                    //记录错误的数据

                }

            }

        });

        int delaySeconds = 2;

        int windowSize = 5;

        //提取EventTime生成WaterMark

        SingleOutputStreamOperator<OrderMain> orderMainStreamWithWaterMark = orderMainDataStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<OrderMain>(Time.seconds(delaySeconds)) {

            @Override

            public long extractTimestamp(OrderMain element) {

                return element.getCreate_time().getTime();

            }

        });

        SingleOutputStreamOperator<OrderDetail> orderDetailStreamWithWaterMark = orderDetailDataStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<OrderDetail>(Time.seconds(delaySeconds)) {

            @Override

            public long extractTimestamp(OrderDetail element) {

                return element.getCreate_time().getTime();

            }

        });

        //定义迟到侧流输出的Tag

        OutputTag<OrderDetail> lateTag = new OutputTag<OrderDetail>("late-date") {};

        //对左表进行单独划分窗口，窗口的长度与cogroup的窗口长度一样

        SingleOutputStreamOperator<OrderDetail> orderDetailWithWindow = orderDetailStreamWithWaterMark.windowAll(TumblingEventTimeWindows.of(Time.seconds(windowSize)))

                .sideOutputLateData(lateTag) //将迟到的数据打上Tag

                .apply(new AllWindowFunction<OrderDetail, OrderDetail, TimeWindow>() {

                    @Override

                    public void apply(TimeWindow window, Iterable<OrderDetail> values, Collector<OrderDetail> out) throws Exception {

                        for (OrderDetail value : values) {

                            out.collect(value);

                        }

                    }

                });

        //获取迟到的数据

        DataStream<OrderDetail> lateOrderDetailStream = orderDetailWithWindow.getSideOutput(lateTag);

        //应为orderDetail表的数据迟到数据不是很多，没必要使用异步IO，直接使用RichMapFunction

        SingleOutputStreamOperator<Tuple2<OrderDetail, OrderMain>> lateOrderDetailAndOrderMain = lateOrderDetailStream.map(new RichMapFunction<OrderDetail, Tuple2<OrderDetail, OrderMain>>() {

            @Override

            public Tuple2<OrderDetail, OrderMain> map(OrderDetail detail) throws Exception {

                return Tuple2.of(detail, null);

            }

        });

        //Left Out JOIN，并且将订单明细表作为左表

        DataStream<Tuple2<OrderDetail, OrderMain>> joined = orderDetailWithWindow.coGroup(orderMainStreamWithWaterMark)

                .where(new KeySelector<OrderDetail, Long>() {

                    @Override

                    public Long getKey(OrderDetail value) throws Exception {

                        return value.getOrder_id();

                    }

                })

                .equalTo(new KeySelector<OrderMain, Long>() {

                    @Override

                    public Long getKey(OrderMain value) throws Exception {

                        return value.getOid();

                    }

                })

                .window(TumblingEventTimeWindows.of(Time.seconds(windowSize)))

                .apply(new CoGroupFunction<OrderDetail, OrderMain, Tuple2<OrderDetail, OrderMain>>() {

                    @Override

                    public void coGroup(Iterable<OrderDetail> first, Iterable<OrderMain> second, Collector<Tuple2<OrderDetail, OrderMain>> out) throws Exception {

                        for (OrderDetail orderDetail : first) {

                            boolean isJoined = false;

                            for (OrderMain orderMain : second) {

                                out.collect(Tuple2.of(orderDetail, orderMain));

                                isJoined = true;

                            }

                            if (!isJoined) {

                                out.collect(Tuple2.of(orderDetail, null));

                            }

                        }

                    }

                });

        joined.union(lateOrderDetailAndOrderMain).map(new RichMapFunction<Tuple2<OrderDetail, OrderMain>, Tuple2<OrderDetail, OrderMain>>() {

            private transient Connection connection;

            @Override

            public void open(Configuration parameters) throws Exception {

                //可以创建数据库连接

                connection = DriverManager.getConnection("jdbc:mysql://172.16.100.100:3306/doit?characterEncoding=UTF-8", "root", "123456");

            }

            @Override

            public Tuple2<OrderDetail, OrderMain> map(Tuple2<OrderDetail, OrderMain> tp) throws Exception {

                //每个关联上订单主表的数据，就查询书库

                if (tp.f1 == null) {

                    tp.f1 = queryOrderMainFromMySQL(tp.f0.getOrder_id(), connection);

                }

                return tp;

            }

            @Override

            public void close() throws Exception {

                //关闭数据库连接

            }

        }).print();

        FlinkUtilsV2.getEnv().execute();

    }

    private static OrderMain queryOrderMainFromMySQL(Long order_id, Connection connection) throws Exception {

        PreparedStatement preparedStatement = connection.prepareStatement("SELECT * ordermain WHERE oid = ?");

        //设置参数

        preparedStatement.setLong(1, order_id);

        //执行查询

        ResultSet resultSet = preparedStatement.executeQuery();

        //取出结果

        long oid = resultSet.getLong("oid");

        Date createTime = resultSet.getDate("create_time");

        double totalMoney = resultSet.getDouble("total_money");

        int status = resultSet.getInt("status");

        OrderMain orderMain = new OrderMain();

        orderMain.setOid(oid);

        orderMain.setStatus(status);

        return orderMain;

    }

}

DataStream<Integer> input = ...;

final OutputTag<String> outputTag = new OutputTag<String>("side-output"){};

SingleOutputStreamOperator<Integer> mainDataStream = input

  .process(new ProcessFunction<Integer, Integer>() {

      @Override

      public void processElement(

          Integer value,

          Context ctx,

          Collector<Integer> out) throws Exception {

        // emit data to regular output

        out.collect(value);

        // emit data to side output

        ctx.output(outputTag, "sideout-" + String.valueOf(value));

      }

    });

package com.wedoctor.flink;

import org.apache.flink.api.java.tuple.Tuple3;

import org.apache.flink.configuration.Configuration;

import org.apache.flink.streaming.api.datastream.DataStream;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.functions.ProcessFunction;

import org.apache.flink.util.Collector;

import org.apache.flink.util.OutputTag;

/**

 * 1.将数据整理成Tuple3

 * 2.然后使用侧流输出将数据分类

 */

public class SideOutputsDemo {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

//        view,pid,2020-03-09 11:42:30

//        activity,a10,2020-03-09 11:42:38

//        order,o345,2020-03-09 11:42:38

        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);

        OutputTag<Tuple3<String, String, String>> viewTag = new OutputTag<Tuple3<String, String, String>>("view-tag") {

        };

        OutputTag<Tuple3<String, String, String>> activityTag = new OutputTag<Tuple3<String, String, String>>("activity-tag") {

        };

        OutputTag<Tuple3<String, String, String>> orderTag = new OutputTag<Tuple3<String, String, String>>("order-tag") {

        };

        //直接调用process方法

        SingleOutputStreamOperator<Tuple3<String, String, String>> tpDataStream = lines.process(new ProcessFunction<String, Tuple3<String, String, String>>() {

            @Override

            public void open(Configuration parameters) throws Exception {

                super.open(parameters);

            }

            @Override

            public void processElement(String input, Context ctx, Collector<Tuple3<String, String, String>> out) throws Exception {

                //1.将字符串转成Tuple2

                String[] fields = input.split(",");

                String type = fields[0];

                String id = fields[1];

                String time = fields[2];

                Tuple3<String, String, String> tp = Tuple3.of(type, id, time);

                //2.对数据打标签

                //将数据打上标签

                if (type.equals("view")) {

                    //输出数据，将数据和标签关联

                    ctx.output(viewTag, tp);  //ctx.output  输出侧流的

                } else if (type.equals("activity")) {

                    ctx.output(activityTag, tp);

                } else {

                    ctx.output(orderTag, tp);

                }

                //输出主流的数据

                out.collect(tp);

            }

        });

        //输出的测流只能通过getSideOutput

        DataStream<Tuple3<String, String, String>> viewDataStream = tpDataStream.getSideOutput(viewTag);

        //分别处理各种类型的数据。

        viewDataStream.print();

        env.execute();

    }

}

package com.wedoctor.flink;

import org.apache.flink.api.common.functions.MapFunction;

import org.apache.flink.api.java.tuple.Tuple;

import org.apache.flink.api.java.tuple.Tuple2;

import org.apache.flink.streaming.api.TimeCharacteristic;

import org.apache.flink.streaming.api.datastream.DataStream;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

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.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;

import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;

import org.apache.flink.streaming.api.windowing.time.Time;

import org.apache.flink.util.OutputTag;

public class WindowLateDataDemo {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //1000,hello

        //2000,jerry

        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);

        //设置了窗口的延迟时间为2秒

        SingleOutputStreamOperator<String> linesWithWaterMark = lines.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(2)) {

            @Override

            public long extractTimestamp(String element) {

                return Long.parseLong(element.split(",")[0]);

            }

        });

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = linesWithWaterMark.map(new MapFunction<String, Tuple2<String, Integer>>() {

            @Override

            public Tuple2<String, Integer> map(String value) throws Exception {

                //切除时间字段，保留单词

                return Tuple2.of(value.split(",")[1], 1);

            }

        });

        KeyedStream<Tuple2<String, Integer>, Tuple> keyed = wordAndOne.keyBy(0);

        OutputTag<Tuple2<String, Integer>> lateDataTag = new OutputTag<Tuple2<String, Integer>>("late-data"){};

        SingleOutputStreamOperator<Tuple2<String, Integer>> summed = keyed.window(TumblingEventTimeWindows.of(Time.seconds(5)))

                .sideOutputLateData(lateDataTag)

                .sum(1);

        //获取迟到数据的侧流

        DataStream<Tuple2<String, Integer>> lateDataStream = summed.getSideOutput(lateDataTag);

        //summed.print("准时的数据: ");

        //lateDataStream.print("迟到的数据：");

        SingleOutputStreamOperator<Tuple2<String, Integer>> result = summed.union(lateDataStream).keyBy(0).sum(1);

        result.print();

        env.execute();

    }

}

// This example implements the asynchronous request and callback with Futures that have the

// interface of Java 8's futures (which is the same one followed by Flink's Future)

/**

 * An implementation of the 'AsyncFunction' that sends requests and sets the callback.

 */

class AsyncDatabaseRequest extends RichAsyncFunction<String, Tuple2<String, String>> {

    /** The database specific client that can issue concurrent requests with callbacks */

    private transient DatabaseClient client;

    @Override

    public void open(Configuration parameters) throws Exception {

        client = new DatabaseClient(host, post, credentials);

    }

    @Override

    public void close() throws Exception {

        client.close();

    }

    @Override

    public void asyncInvoke(String key, final ResultFuture<Tuple2<String, String>> resultFuture) throws Exception {

        // issue the asynchronous request, receive a future for result

        final Future<String> result = client.query(key);

        // set the callback to be executed once the request by the client is complete

        // the callback simply forwards the result to the result future

        CompletableFuture.supplyAsync(new Supplier<String>() {

            @Override

            public String get() {

                try {

                    return result.get();

                } catch (InterruptedException | ExecutionException e) {

                    // Normally handled explicitly.

                    return null;

                }

            }

        }).thenAccept( (String dbResult) -> {

            resultFuture.complete(Collections.singleton(new Tuple2<>(key, dbResult)));

        });

    }

}

// create the original stream

DataStream<String> stream = ...;

// apply the async I/O transformation

DataStream<Tuple2<String, String>> resultStream =

    AsyncDataStream.unorderedWait(stream, new AsyncDatabaseRequest(), 1000, TimeUnit.MILLISECONDS, 100);

<!--发送HTTP请求的Java工具包 -->

        <dependency>

            <groupId>org.apache.httpcomponents</groupId>

            <artifactId>httpclient</artifactId>

            <version>4.5.7</version>

        </dependency>

        <!-- 高效的异步HttpClient -->

        <dependency>

            <groupId>org.apache.httpcomponents</groupId>

            <artifactId>httpasyncclient</artifactId>

            <version>4.1.4</version>

        </dependency>

        <!-- 同步的HTTP Client -->

        <dependency>

            <groupId>commons-httpclient</groupId>

            <artifactId>commons-httpclient</artifactId>

            <version>3.1</version>

        </dependency>

        <dependency>

            <groupId>com.alibaba</groupId>

            <artifactId>fastjson</artifactId>

            <version>1.2.57</version>

        </dependency>

<!--发送HTTP请求的Java工具包 -->

        <dependency>

            <groupId>org.apache.httpcomponents</groupId>

            <artifactId>httpclient</artifactId>

            <version>4.5.7</version>

        </dependency>

        <!-- 高效的异步HttpClient -->

        <dependency>

            <groupId>org.apache.httpcomponents</groupId>

            <artifactId>httpasyncclient</artifactId>

            <version>4.1.4</version>

        </dependency>

        <!-- 同步的HTTP Client -->

        <dependency>

            <groupId>commons-httpclient</groupId>

            <artifactId>commons-httpclient</artifactId>

            <version>3.1</version>

        </dependency>

        <dependency>

            <groupId>com.alibaba</groupId>

            <artifactId>fastjson</artifactId>

            <version>1.2.57</version>

        </dependency>

<dependency>

            <groupId>com.alibaba</groupId>

            <artifactId>druid</artifactId>

            <version>1.0.18</version>

        </dependency>

package com.wedoctor.flink;

import com.alibaba.druid.pool.DruidDataSource;

import org.apache.flink.configuration.Configuration;

import org.apache.flink.streaming.api.functions.async.ResultFuture;

import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.sql.Connection;

import java.sql.PreparedStatement;

import java.sql.ResultSet;

import java.sql.SQLException;

import java.util.Collections;

import java.util.concurrent.CompletableFuture;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.Future;

import java.util.function.Supplier;

public class AsyncMysqlRequest extends RichAsyncFunction<String,String> {

    private transient DruidDataSource dataSource;

    private transient ExecutorService executorService;

    @Override

    public void open(Configuration parameters) throws Exception {

        super.open(parameters);

        executorService = Executors.newFixedThreadPool(20);

        dataSource = new DruidDataSource();

        dataSource.setDriverClassName("com.mysql.jdbc.Driver");

        dataSource.setUsername("root");

        dataSource.setPassword("123456");

        dataSource.setUrl("jdbc:mysql://localhost:3306/bigdata?characterEncoding=UTF-8");

        dataSource.setInitialSize(5);

        dataSource.setMinIdle(10);

        dataSource.setMaxActive(20);

    }

    @Override

    public void close() throws Exception {

        super.close();

        dataSource.close();

        executorService.shutdown();

    }

    @Override

    public void asyncInvoke(String id, ResultFuture<String> resultFuture) throws Exception {

        Future<String> future = executorService.submit(() -> {

            return queryFromMysql(id);

        });

        CompletableFuture.supplyAsync(new Supplier<String>() {

            @Override

            public String get() {

                try {

                    return future.get();

                } catch (Exception e) {

                    return null;

                }

            }

        }).thenAccept( (String dbResult) -> {

            resultFuture.complete(Collections.singleton(dbResult));

        });

    }

    private String queryFromMysql(String param) throws SQLException {

        String sql = "select name from info where id = ?";

        String result = null;

        Connection connection = null;

        PreparedStatement stmt = null;

        ResultSet rs = null;

        try {

            connection = dataSource.getConnection();

            stmt = connection.prepareStatement(sql);

            stmt.setString(1,param);

            rs = stmt.executeQuery();

            while (rs.next()){

                result = rs.getString("name");

            }

        }finally {

            if (rs != null){

                rs.close();

            }

            if (stmt != null){

                stmt.close();

            }

            if (connection != null){

                connection.close();

            }

        }

        if (result != null){

            //可以放入缓存中

        }

        return result;

    }

}