大数据流试计算引擎Flink篇

Apache Flink是一个框架和分布式处理引擎，用于在无边界和有边界数据流上进行有状态的计算。

<properties>
    <encoding>UTF-8</encoding>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <maven.compiler.source>1.8</maven.compiler.source>
    <maven.compiler.target>1.8</maven.compiler.target>
    <java.version>1.8</java.version>
    <scala.version>2.12</scala.version>
    <flink.version>1.13.1</flink.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <version>1.18.16</version>
    </dependency>
    <!--flink客户端-->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients_${scala.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <!--scala版本-->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-scala_${scala.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <!--flink-java版本-->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <!--streamingጱscala版本-->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-scala_${scala.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <!--streamingጱjava版本-->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_${scala.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <!--日志输出-->
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <version>1.7.7</version>
        <scope>runtime</scope>
        </dependency>
    <dependency>
        <groupId>log4j</groupId>
        <artifactId>log4j</artifactId>
        <version>1.2.17</version>
        <scope>runtime</scope>
    </dependency>
    <!--json依赖包-->
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>1.2.44</version>
    </dependency>
</dependencies>

新建log4j.properties，日志文件
### 配置appender名称
log4j.rootLogger = debugFile, errorFile
### debug级别以上的日志到：src/logs/debug.log
log4j.appender.debugFile = org.apache.log4j.DailyRollingFileAppender
log4j.appender.debugFile.File = src/logs/flink.log
log4j.appender.debugFile.Append = true
### Threshold属性指定输出等级
log4j.appender.debugFile.Threshold = info
log4j.appender.debugFile.layout = org.apache.log4j.PatternLayout
log4j.appender.debugFile.layout.ConversionPattern = %-d{yyyy-MM-dd HH:mm:ss}  [ %t:%r ] - [ %p ]  %n%m%n
### error级别以上的日志 src/logs/error.log
log4j.appender.errorFile = org.apache.log4j.DailyRollingFileAppender
log4j.appender.errorFile.File = src/logs/error.log
log4j.appender.errorFile.Append = true
log4j.appender.errorFile.Threshold = error
log4j.appender.errorFile.layout = org.apache.log4j.PatternLayout
log4j.appender.errorFile.layout.ConversionPattern = %-d{yyyy-MM-dd HH:mm:ss}  [ %t:%r ] - [ %p ]  %n%m%n

/**
 * @author lixiang
 */
public class Test {
    private static List<String> list = new ArrayList<>();

    static {
        list.add("SpringBoot,Docker");
        list.add("Netty,SpringCloud");
        list.add("Flink,Linux");
    }

    public static void test1() {
        Tuple3<Integer, String, Integer> tuple3 = Tuple3.of(1, "lixiang", 23);
        System.out.println(tuple3.f0);
        System.out.println(tuple3.f1);
        System.out.println(tuple3.f2);
    }

    public static void test2() {
        List<String> collect = list.stream().map(obj -> obj + "拼接").collect(Collectors.toList());
        System.out.println(collect);
    }

    public static void test3() {
        List<String> collect = list.stream().flatMap(obj -> Arrays.stream(obj.split(","))).collect(Collectors.toList());
        System.out.println(collect);
    }
}

public static void main(String[] args) {
     test1();
     test2();
     test3();
}

/**
 * @author lixiang
 */
public class FlinkDemo {
    public static void main(String[] args) throws Exception {
        //构建执行任务环境以及任务的启动入口，存储全局相关的参数
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //设置并行度
        env.setParallelism(1);

        //相同类型元素的数据流
        DataStream<String> stringDataStream = env.fromElements("Java,SpringBoot,SpringCloud", "Java,Linux,Docker");

        stringDataStream.print("处理前");

        //FlatMapFunction<Strings,String>，key是输入类型，value是Collector响应的收集的类型，看源码的注释也是DataStream<String>里面泛型类型
        DataStream<String> flatMapDataStream = stringDataStream.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                String[] arr = value.split(",");
                for (String s : arr) {
                    out.collect(s);
                }
            }
        });
        flatMapDataStream.print("处理后");

        //DataStream需要execute，可以取个名称
        env.execute("data stream job");
    }
}

//设置并行度
env.setParallelism(3);

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

<!--Flink web ui-->
<dependency>
     <groupId>org.apache.flink</groupId>
     <artifactId>flink-runtime-web_${scala.version}</artifactId>
     <version>${flink.version}</version>
</dependency>

/**
 * flink UI demo
 * @author lixiang
 */
public class FlinkUI {

    public static void main(String[] args) throws Exception {
        //构建执行任务环境以及任务的启动入口
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        //设置并行度
        env.setParallelism(1);
        //监听192.168.139.80:8888输送过来的数据流
        DataStreamSource<String> stream = env.socketTextStream("192.168.139.80", 8888);
        //流处理
        SingleOutputStreamOperator<String> streamOperator = stream.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                String[] split = value.split(",");
                for (String s : split) {
                    out.collect(s);
                }
            }
        });
        streamOperator.print("处理后");
        //执行任务
        env.execute("data stream job");
    }

}

env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

/**
 * @author lixiang
 */
public class FlinkSourceDemo {
    public static void main(String[] args) throws Exception {
        //构建执行任务环境以及任务的启动入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> ds1 = env.fromElements("java,springboot", "kafka,redis", "openstack,k8s,docker");
        ds1.print("ds1");

        DataStreamSource<String> ds2 = env.fromCollection(Arrays.asList("hive", "hadoop", "hbase", "rabbitmq", "java"));
        ds2.print("ds2");

        DataStreamSource<Long> ds3 = env.fromSequence(0, 10);
        ds3.print("ds3");

        //执行任务
        env.execute("data job");
    }
}

/**
 * @author lixiang
 */
public class FlinkSourceDemo2 {
    public static void main(String[] args) throws Exception {
        //构建执行任务环境以及任务的启动的入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStream<String> textDS = env.readTextFile("E:\\软件资料\\log.txt");
        //DataStream<String> hdfsDS = env.readTextFile("hdfs://lixiang:8010/file/log/words.txt");
        textDS.print("textDS");
        env.execute("text job");
    }
}

/**
 * flink UI demo
 * @author lixiang
 */
public class FlinkUI {

    public static void main(String[] args) throws Exception {
        //构建执行任务环境以及任务的启动入口
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        //设置并行度
        env.setParallelism(1);
        //监听192.168.139.80:8888输送过来的数据流
        DataStreamSource<String> stream = env.socketTextStream("192.168.139.80", 8888);
        //流处理
        SingleOutputStreamOperator<String> streamOperator = stream.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                String[] split = value.split(",");
                for (String s : split) {
                    out.collect(s);
                }
            }
        });
        streamOperator.print("处理后");
        //执行任务
        env.execute("data stream job");
    }

}

/**
 * 订单实体类
 * @author lixiang
 */
@Data
@NoArgsConstructor
@AllArgsConstructor
public class VideoOrder
{
    private String tradeNo;
    private String title;
    private int money;
    private int userId;
    private Date createTime;
}

/**
 * @author lixiang
 */
public class VideoOrderSource extends RichParallelSourceFunction<VideoOrder>
{

    private volatile Boolean flag = true;

    private Random random = new Random();

    private static List<String> list = new ArrayList<>();

    static
    {
        list.add("SpringBoot2.x课程");
        list.add("Linux入到到精通");
        list.add("Flink流式技术课程");
        list.add("Kafka流式处理消息平台");
        list.add("微服务SpringCloud教程");
    }

    @Override
    public void run(SourceContext<VideoOrder> sourceContext) throws Exception {
        int x = 0;
        while (flag)
        {
            Thread.sleep(1000);
            String id = UUID.randomUUID().toString();
            int userId = random.nextInt(10);
            int money = random.nextInt(100);
            int videoNum = random.nextInt(list.size());
            String title = list.get(videoNum);
            sourceContext.collect(new VideoOrder(id, title, money, userId, new Date()));
            x++;
            if (x == 10)
            {
                cancel();
            }
        }
    }

    /**
     * 取消任务
     */
    @Override
    public void cancel()
    {
        flag = false;
    }
}

/**
 * @author lixiang
 */
public class FlinkMainSource {
    public static void main(String[] args) throws Exception {
        //构建执行任务环境以及任务的启动入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStreamSource<VideoOrder> source = env.addSource(new VideoOrderSource());

        source.print("接入的数据");

        SingleOutputStreamOperator<Integer> streamOperator = source.flatMap(new FlatMapFunction<VideoOrder, Integer>() {
            @Override
            public void flatMap(VideoOrder value, Collector<Integer> out) throws Exception {
                out.collect(value.getMoney());
            }
        });
        streamOperator.print("处理后");
        //流程启动
        env.execute("custom source job");
    }
}

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

docker pull mysql:5.7
docker run -itd -p 3306:3306 --name my-mysql -e MYSQL_ROOT_PASSWORD=123456 mysql:5.7

CREATE TABLE `video_order` (
  `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
  `user_id` int(11) DEFAULT NULL,
  `money` int(11) DEFAULT NULL,
  `title` varchar(32) DEFAULT NULL,
  `trade_no` varchar(64) DEFAULT NULL,
  `create_time` date DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-jdbc_2.12</artifactId>
    <version>1.12.0</version>
</dependency>

<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>8.0.25</version>
</dependency>

/**
 * @author lixiang
 */
public class MysqlSink extends RichSinkFunction<VideoOrderDO> {

    private Connection conn = null;
    private PreparedStatement ps = null;

    @Override
    public void invoke(VideoOrderDO videoOrder, Context context) throws Exception {
        //给ps中的?设置具体值
        ps.setInt(1, videoOrder.getUserId());
        ps.setInt(2, videoOrder.getMoney());
        ps.setString(3, videoOrder.getTitle());
        ps.setString(4, videoOrder.getTradeNo());
        ps.setDate(5, new Date(videoOrder.getCreateTime().getTime()));
        int i = ps.executeUpdate();
        System.out.println("处理数据，插入数据库结果：" + (i > 0));
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        System.out.println("---open---");
        conn = DriverManager.getConnection("jdbc:mysql://192.168.139.20:3306/flink?useUnicode=true&characterEncoding=utf8&allowMultiQueries=true&serverTimezone=Asia/Shanghai", "root", "123456");
        String sql = "INSERT INTO `video_order` (`user_id`, `money`, `title`, `trade_no`, `create_time`) VALUES(?,?,?,?,?);";
        ps = conn.prepareStatement(sql);
    }

    @Override
    public void close() throws Exception {
        if (conn != null) {
            conn.close();
        }
        if (ps != null) {
            ps.close();
        }
        System.out.println("---close---");
    }
}

public class FlinkMainSink {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);

        DataStreamSource<VideoOrderDO> source = env.addSource(new VideoOrderSource());
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
                3, // 尝试重启的次数
                Time.of(10, TimeUnit.SECONDS) // 间隔
        ));
        source.print("接收的数据");

        source.addSink(new MysqlSink());
        //流程启动
        env.execute("custom sink job");
    }
}

拉取镜像：docker pull redis
启动容器：docker run -d --name redis -p 6379:6379 redis --requirepass "123456"

<dependency>
     <groupId>org.apache.bahir</groupId>
     <artifactId>flink-connector-redis_2.11</artifactId>
     <version>1.0</version>
</dependency>

/**
 * 定义泛型，就是要返回的类型
 * @author lixiang
 */
public class MyRedisSink implements RedisMapper<Tuple2<String,Integer>> {

    /**
     * 选择对应的数据结构，和key的名称
     * @return
     */
    @Override
    public RedisCommandDescription getCommandDescription() {
        return new RedisCommandDescription(RedisCommand.HSET,"VIDEO_ORDER_COUNTER");
    }
    
    /**
     * 返回key
     * @param value
     * @return
     */
    @Override
    public String getKeyFromData(Tuple2<String,Integer> value) {
        return value.f0;
    }

    /**
     * 返回value
     * @param value
     * @return
     */
    @Override
    public String getValueFromData(Tuple2<String,Integer> value) {
        return value.f1.toString();
    }
}

/**
 * @author lixiang
 */
public class FlinkRedisDemo {
    public static void main(String[] args) throws Exception {
        //构建执行任务环境以及任务的启动入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //自己构建的数据源
        /*DataStream<VideoOrderDO> ds = env.fromElements(new VideoOrderDO(5, 32, "java", "2123143432", new Date()),
                new VideoOrderDO(5, 40, "spring", "2123143432", new Date()),
                new VideoOrderDO(5, 60, "springBoot", "2233143432", new Date()),
                new VideoOrderDO(5, 29, "springBoot", "2125643432", new Date()),
                new VideoOrderDO(5, 67, "docker", "2129843432", new Date()),
                new VideoOrderDO(5, 89, "java", "2120943432", new Date()));*/

        //使用自定义的source
        DataStream<VideoOrderDO> ds = env.addSource(new VideoOrderSource());

        //map转换。来一个记录一个，方便后续统计
        DataStream<Tuple2<String,Integer>> mapDS = ds.map(new MapFunction<VideoOrderDO, Tuple2<String,Integer>>() {
            @Override
            public Tuple2<String,Integer> map(VideoOrderDO value) throws Exception {
                return new Tuple2<>(value.getTitle(), 1);
            }
        });

        KeyedStream<Tuple2<String, Integer>, String> keyedStream = mapDS.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
            @Override
            public String getKey(Tuple2<String, Integer> value) throws Exception {
                return value.f0;
            }
        });

        DataStream<Tuple2<String, Integer>> sumDS = keyedStream.sum(1);

        //输出统计
        sumDS.print();

        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("192.168.139.20").setPassword("123456").setPort(6379).build();

        sumDS.addSink(new RedisSink<>(conf,new MyRedisSink()));

        //DataStream需要调用execute，可以取这个名称
        env.execute("custom redis job");
    }
}

/**
 * 自定义的source
 * @author lixiang
 */
public class VideoOrderSource extends RichParallelSourceFunction<VideoOrderDO>
{

    private volatile Boolean flag = true;

    private Random random = new Random();

    private static List<String> list = new ArrayList<>();

    static
    {
        list.add("SpringBoot2.x");
        list.add("Linux");
        list.add("Flink");
        list.add("Kafka");
        list.add("SpringCloud");
        list.add("SpringBoot");
        list.add("Docker");
        list.add("Netty");
    }

    @Override
    public void run(SourceContext<VideoOrderDO> sourceContext) throws Exception {
        int x = 0;
        while (flag)
        {
            Thread.sleep(1000);
            String id = UUID.randomUUID().toString();
            int userId = random.nextInt(10);
            int money = random.nextInt(100);
            int videoNum = random.nextInt(list.size());
            String title = list.get(videoNum);
            String uuid = UUID.randomUUID().toString();
            sourceContext.collect(new VideoOrderDO(userId, money, title,uuid, new Date()));
        }
    }

    /**
     * 取消任务
     */
    @Override
    public void cancel()
    {
        flag = false;
    }
}

#zk部署
docker run -d --name zookeeper -p 2181:2181 -t wurstmeister/zookeeper

#kafka部署，换成自己的IP
docker run -d --name kafka \
-p 9092:9092 \
-e KAFKA_BROKER_ID=0 \
-e KAFKA_ZOOKEEPER_CONNECT=192.168.139.20:2181 \
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.139.20:9092 \
-e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 wurstmeister/kafka

#进入容器内部，创建topic
docker exec -it kafka /bin/bash

cd /opt/kafka
bin/kafka-topics.sh --create --zookeeper 192.168.139.20:2181 --replication-factor 1 --partitions 1 --topic test-topic

#创建生产者发送消息
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test-topic

#运行一个消费者，注意--from-beginning从开头第一个开始消费
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic --from-beginning

<dependency>
     <groupId>org.apache.flink</groupId>
     <artifactId>flink-connector-kafka_${scala.version}</artifactId>
     <version>${flink.version}</version>
</dependency>

/**
 * @author lixiang
 */
public class FlinkKafka {
    public static void main(String[] args) throws Exception {
        //构建执行任务环境以及任务的启动入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties props = new Properties();
        //kafka地址
        props.setProperty("bootstrap.servers", "192.168.139.20:9092");
        //组名
        props.setProperty("group.id", "video-order-group");
        //字符串序列化和反序列化规则
        props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        //offset重置规则
        props.setProperty("auto.offset.reset", "latest");
        //自动提交
        props.setProperty("enable.auto.commit", "true");
        props.setProperty("auto.commit.interval.ms", "2000");
        //有后台线程每隔10s检测一下Kafka的分区变化情况
        props.setProperty("flink.partition-discovery.interval-millis","10000");

        //监听test-topic发送的消息
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("test-topic",new SimpleStringSchema(),props);

        consumer.setStartFromGroupOffsets();

        DataStream<String> consumerDS = env.addSource(consumer);

        consumerDS.print("test-topic接收的消息");

        //接到消息后，处理
        DataStream<String> dataStream = consumerDS.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                return "新来一个订单课程："+value;
            }
        });
		//处理后的消息发送到order-topic
        FlinkKafkaProducer<String> producer = new FlinkKafkaProducer<>("order-topic",new SimpleStringSchema(),props);
        dataStream.addSink(producer);
        env.execute("kafka job");
    }
}

#创建生产者发送消息
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test-topic

#运行一个消费者
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic order-topic --from-beginning

/**
 * @author lixiang
 * flink map算子demo
 */
public class FlinkMapDemo {
    public static void main(String[] args) throws Exception {
        //构建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStream<VideoOrderDO> streamSource = env.addSource(new VideoOrderSource());
        streamSource.print("处理前数据");

        DataStream<Tuple2<String,Integer>> dataStream = streamSource.map(new MapFunction<VideoOrderDO, Tuple2<String,Integer>>() {
            @Override
            public Tuple2<String, Integer> map(VideoOrderDO value) throws Exception {
                return new Tuple2<>(value.getTitle(),value.getMoney());
            }
        });

        dataStream.print("处理后");

        env.execute("map job");
    }
}

/**
 * @author lixiang
 * flatMap 算子demo
 */
public class FlinkFlatMapDemo {
    public static void main(String[] args) throws Exception {
        //构建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStream<String> ds = env.fromElements("java&35,spring&20,springboot&30", "springcloud&21,shiro&39,docker&56,linux&87", "netty&98,kafka&48");
        ds.print("处理前");
        DataStream<Tuple2<String,Integer>> out = ds.flatMap(new FlatMapFunction<String, Tuple2<String,Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] element = value.split(",");
                for (String s : element) {
                    String[] eles = s.split("&");
                    out.collect(new Tuple2<>(eles[0],Integer.parseInt(eles[1])));
                }
            }
        });

        out.print("处理后");

        env.execute("flatMap job");

    }
}

/**
 * @author lixiang
 * flink map算子demo
 */
public class FlinkMapDemo {
    public static void main(String[] args) throws Exception {
        //构建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStream<VideoOrderDO> streamSource = env.addSource(new VideoOrderSource());
        streamSource.print("处理前数据");

        DataStream<Tuple2<String,Integer>> dataStream = streamSource.map(new RichMapFunction<VideoOrderDO, Tuple2<String,Integer>>() {
            @Override
            public Tuple2<String, Integer> map(VideoOrderDO value) throws Exception {
                return new Tuple2<>(value.getTitle(),value.getMoney());
            }

            @Override
            public void open(Configuration parameters) throws Exception {
                System.out.println("open方法执行");
            }

            @Override
            public void close() throws Exception {
                System.out.println("close方法执行");
            }
        });

        dataStream.print("处理后");

        env.execute("map job");
    }
}

/**
 * @author lixiang
 * flatMap 算子demo
 */
public class FlinkFlatMapDemo {
    public static void main(String[] args) throws Exception {
        //构建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStream<String> ds = env.fromElements("java&35,spring&20,springboot&30", "springcloud&21,shiro&39,docker&56,linux&87", "netty&98,kafka&48");
        ds.print("处理前");
        DataStream<Tuple2<String,Integer>> out = ds.flatMap(new RichFlatMapFunction<String, Tuple2<String,Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] element = value.split(",");
                for (String s : element) {
                    String[] eles = s.split("&");
                    out.collect(new Tuple2<>(eles[0],Integer.parseInt(eles[1])));
                }
            }

            @Override
            public void open(Configuration parameters) throws Exception {
                System.out.println("open方法执行");
            }

            @Override
            public void close() throws Exception {
                System.out.println("close方法执行");
            }
        });

        out.print("处理后");

        env.execute("flatMap job");

    }
}

/**
 * @author lixiang
 * keyBy 算子demo
 */
public class FlinkKeyByDemo {
    public static void main(String[] args) throws Exception {
        //构建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        DataStream<VideoOrderDO> dataStream = env.addSource(new VideoOrderSource());

        //根据title进行分组
        KeyedStream<VideoOrderDO,String> keyedStream = dataStream.keyBy(new KeySelector<VideoOrderDO, String>() {
            @Override
            public String getKey(VideoOrderDO value) throws Exception {
                return value.getTitle();
            }
        });

        //分组后将相同标题的money进行累加
        SingleOutputStreamOperator<VideoOrderDO> sumDS = keyedStream.sum("money");

        //map转换
        DataStream<Tuple2<String, Integer>> outputStreamOperator = sumDS.map(new MapFunction<VideoOrderDO, Tuple2<String,Integer>>() {
            @Override
            public Tuple2<String, Integer> map(VideoOrderDO value) throws Exception {
                return new Tuple2<>(value.getTitle(),value.getMoney());
            }
        });

        outputStreamOperator.print();

        env.execute("keyBy job");
    }
}

/**
 * @author lixiang
 * flink filter算子demo
 * 先过滤money大于30的，然后根据标题进行分组，然后求每组money总和，最后map转换
 */
public class FlinkFliterDemo {
    public static void main(String[] args) throws Exception {
        //构建执行任务环境以及任务的启动的入口, 存储全局相关的参数
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        env.setParallelism(1);

        DataStreamSource<VideoOrderDO> ds = env.addSource(new VideoOrderSource());

        DataStream<Tuple2<String,Integer>> out = ds.filter(new FilterFunction<VideoOrderDO>() {
            @Override
            public boolean filter(VideoOrderDO value) throws Exception {
                return value.getMoney()>30;
            }
        }).keyBy(new KeySelector<VideoOrderDO, String>() {
            @Override
            public String getKey(VideoOrderDO value) throws Exception {
                return value.getTitle();
            }
        }).sum("money").map(new MapFunction<VideoOrderDO, Tuple2<String,Integer>>() {
            @Override
            public Tuple2<String, Integer> map(VideoOrderDO value) throws Exception {
                return new Tuple2<>(value.getTitle(), value.getMoney());
            }
        });

        out.print();

        env.execute("filter sum job");
    }
}

/**
 * @author lixiang
 * reduce 算子demo
 */
public class FlinkReduceDemo {
    public static void main(String[] args) throws Exception {
        //构建执行任务环境以及任务的启动的入口, 存储全局相关的参数
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        env.setParallelism(1);

        DataStreamSource<VideoOrderDO> ds = env.addSource(new VideoOrderSource());

        SingleOutputStreamOperator<Tuple2<String,Integer>> reduce = ds.keyBy(new KeySelector<VideoOrderDO, String>() {
            @Override
            public String getKey(VideoOrderDO value) throws Exception {
                return value.getTitle();
            }
        }).reduce(new AggregationFunction<VideoOrderDO>() {
            //value1是历史对象，value2是加入统计的对象，所以value1.f1是历史值，value2.f1是新值，不断累加
            @Override
            public VideoOrderDO reduce(VideoOrderDO value1, VideoOrderDO value2) throws Exception {
                value1.setMoney(value1.getMoney() + value2.getMoney());
                return value1;
            }
        }).map(new MapFunction<VideoOrderDO, Tuple2<String,Integer>>() {
            @Override
            public Tuple2<String,Integer> map(VideoOrderDO value) throws Exception {
                return new Tuple2<>(value.getTitle(),value.getMoney());
            }
        });

        reduce.print();

        env.execute("reduce job");
    }
}

/**
 * @author lixiang
 * maxBy-max-minBy-min的使用
 */
public class FlinkMinMaxDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        //env.setParallelism(1);
        DataStream<VideoOrderDO> ds = env.fromElements(new VideoOrderDO(5, 32, "java", "2123143432", new Date()),
                new VideoOrderDO(25, 40, "spring", "2123143432", new Date()),
                new VideoOrderDO(45, 60, "springBoot", "2233143432", new Date()),
                new VideoOrderDO(15, 29, "springBoot", "2125643432", new Date()),
                new VideoOrderDO(54, 67, "java", "2129843432", new Date()),
                new VideoOrderDO(59, 89, "java", "2120943432", new Date()));

        SingleOutputStreamOperator<VideoOrderDO> out = ds.keyBy(new KeySelector<VideoOrderDO, String>() {
            @Override
            public String getKey(VideoOrderDO value) throws Exception {
                return value.getTitle();
            }
        }).max("money");

        out.print();

        env.execute("max job");
    }
}

/**
 * @author lixiang
 * maxBy-max-minBy-min的使用
 */
public class FlinkMinMaxDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        //env.setParallelism(1);
        DataStream<VideoOrderDO> ds = env.fromElements(new VideoOrderDO(5, 32, "java", "2123143432", new Date()),
                new VideoOrderDO(25, 40, "spring", "2123143432", new Date()),
                new VideoOrderDO(45, 60, "springBoot", "2233143432", new Date()),
                new VideoOrderDO(15, 29, "springBoot", "2125643432", new Date()),
                new VideoOrderDO(54, 67, "java", "2129843432", new Date()),
                new VideoOrderDO(59, 89, "java", "2120943432", new Date()));

        SingleOutputStreamOperator<VideoOrderDO> out = ds.keyBy(new KeySelector<VideoOrderDO, String>() {
            @Override
            public String getKey(VideoOrderDO value) throws Exception {
                return value.getTitle();
            }
        }).maxBy("money");

        out.print();

        env.execute("max job");
    }
}

aggregate(agg函数,WindowFunction(){  })

AggregateFunction<IN, ACC, OUT>
IN是输入类型，ACC是中间聚合状态类型，OUT是输出类型，是聚合统计当前窗口的数据

apply(new processWindowFunction(){})

IN是输入类型，OUT是输出类型，KEY是分组类型，W是时间窗

WindowFunction<IN, OUT, KEY, W extends Window>

//对数据进行解析 ,process对每个元素进行处理，相当于 map+flatMap+filter

process(new KeyedProcessFunction(){processElement、onTimer})

/**
 * @author lixiang
 * Tumbling-Window滚动窗口
 */
public class FlinkTumblingDemo {
    public static void main(String[] args) throws Exception {
        //构建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        DataStream<VideoOrderDO> ds = env.addSource(new VideoOrderSource());

        KeyedStream<VideoOrderDO, String> keyedStream = ds.keyBy(new KeySelector<VideoOrderDO, String>() {
            @Override
            public String getKey(VideoOrderDO value) throws Exception {
                return value.getTitle();
            }
        });
        SingleOutputStreamOperator<Map<String, Object>> map = keyedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(5))).sum("money").map(new MapFunction<VideoOrderDO, Map<String, Object>>() {
            @Override
            public Map<String, Object> map(VideoOrderDO value) throws Exception {
                Map<String, Object> map = new HashMap<>();
                map.put("title", value.getTitle());
                map.put("money", value.getMoney());
                map.put("createDate", TimeUtil.toDate(value.getCreateTime()));
                return map;
            }
        });

        map.print();
        env.execute("Tumbling Window job");

    }
}

/**
 * @author lixiang
 * Tumbling-Window滚动窗口
 */
public class FlinkSlidingDemo {
    public static void main(String[] args) throws Exception {
        //构建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        DataStream<VideoOrderDO> ds = env.addSource(new VideoOrderSource());

        KeyedStream<VideoOrderDO, String> keyedStream = ds.keyBy(new KeySelector<VideoOrderDO, String>() {
            @Override
            public String getKey(VideoOrderDO value) throws Exception {
                return value.getTitle();
            }
        });
        //每5s去统计过去20s的数据
        SingleOutputStreamOperator<Map<String, Object>> map = keyedStream.window(SlidingProcessingTimeWindows.of(Time.seconds(20),Time.seconds(5))).sum("money").map(new MapFunction<VideoOrderDO, Map<String, Object>>() {
            @Override
            public Map<String, Object> map(VideoOrderDO value) throws Exception {
                Map<String, Object> map = new HashMap<>();
                map.put("title", value.getTitle());
                map.put("money", value.getMoney());
                map.put("createDate", TimeUtil.toDate(value.getCreateTime()));
                return map;
            }
        });

        map.print();
        env.execute("Sliding Window job");
    }
}

/**
 * @author lixiang
 * Tumbling-Window滚动窗口
 */
public class FlinkWindow1Demo {
    public static void main(String[] args) throws Exception {
        //构建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        DataStream<VideoOrderDO> ds = env.addSource(new VideoOrderSource());

        KeyedStream<VideoOrderDO, String> keyedStream = ds.keyBy(new KeySelector<VideoOrderDO, String>() {
            @Override
            public String getKey(VideoOrderDO value) throws Exception {
                return value.getTitle();
            }
        });
        SingleOutputStreamOperator<Map<String, Object>> map = keyedStream.countWindow(5).sum("money").map(new MapFunction<VideoOrderDO, Map<String, Object>>() {
            @Override
            public Map<String, Object> map(VideoOrderDO value) throws Exception {
                Map<String, Object> map = new HashMap<>();
                map.put("title", value.getTitle());
                map.put("money", value.getMoney());
                map.put("createDate", TimeUtil.toDate(value.getCreateTime()));
                return map;
            }
        });

        map.print();

        env.execute("Count Window job");

    }
}

/**
 * @author lixiang
 * Tumbling-Window滚动窗口
 */
public class FlinkWindow1Demo {
    public static void main(String[] args) throws Exception {
        //构建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        DataStream<VideoOrderDO> ds = env.addSource(new VideoOrderSource());

        KeyedStream<VideoOrderDO, String> keyedStream = ds.keyBy(new KeySelector<VideoOrderDO, String>() {
            @Override
            public String getKey(VideoOrderDO value) throws Exception {
                return value.getTitle();
            }
        });
        SingleOutputStreamOperator<Map<String, Object>> map = keyedStream.countWindow(5,2).sum("money").map(new MapFunction<VideoOrderDO, Map<String, Object>>() {
            @Override
            public Map<String, Object> map(VideoOrderDO value) throws Exception {
                Map<String, Object> map = new HashMap<>();
                map.put("title", value.getTitle());
                map.put("money", value.getMoney());
                map.put("createDate", TimeUtil.toDate(value.getCreateTime()));
                return map;
            }
        });

        map.print();

        env.execute("Count Window job");

    }
}

aggregate(agg函数,WindowFunction(){  })

AggregateFunction<IN, ACC, OUT>
IN是输入类型，ACC是中间聚合状态类型，OUT是输出类型，是聚合统计当前窗口的数据

/**
 * @author lixiang
 * Tumbling-Window滚动窗口
 */
public class FlinkWindow1Demo {
    public static void main(String[] args) throws Exception {
        //构建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        DataStream<VideoOrderDO> ds = env.addSource(new VideoOrderSource());

        WindowedStream<VideoOrderDO, String, TimeWindow> stream = ds.keyBy(new KeySelector<VideoOrderDO, String>() {
            @Override
            public String getKey(VideoOrderDO value) throws Exception {
                return value.getTitle();
            }
        }).window(TumblingProcessingTimeWindows.of(Time.seconds(5)));

        stream.aggregate(new AggregateFunction<VideoOrderDO, Map<String, Object>, Map<String, Object>>() {

            //初始化累加器
            @Override
            public Map<String, Object> createAccumulator() {
                return new HashMap<>();
            }

            //聚合方式
            @Override
            public Map<String, Object> add(VideoOrderDO value, Map<String, Object> accumulator) {
                if (accumulator.size() == 0) {
                    accumulator.put("title", value.getTitle());
                    accumulator.put("money", value.getMoney());
                    accumulator.put("num", 1);
                    accumulator.put("createTime", value.getCreateTime());
                } else {
                    accumulator.put("title", value.getTitle());
                    accumulator.put("money", value.getMoney() + Integer.parseInt(accumulator.get("money").toString()));
                    accumulator.put("num", 1 + Integer.parseInt(accumulator.get("num").toString()));
                    accumulator.put("createTime", value.getCreateTime());
                }

                return accumulator;
            }

            //返回结果
            @Override
            public Map<String, Object> getResult(Map<String, Object> accumulator) {
                return accumulator;
            }
            //合并内容
            @Override
            public Map<String, Object> merge(Map<String, Object> a, Map<String, Object> b) {
                return null;
            }
        }).print();
        env.execute("Tumbling Window job");
    }
}

apply(new WindowFunction(){ })

IN是输入类型，OUT是输出类型，KEY是分组类型，W是时间窗 
WindowFunction<IN, OUT, KEY, W extends Window>

/**
 * @author lixiang
 * apply
 */
public class FlinkWindow2Demo {
    public static void main(String[] args) throws Exception {
        //构建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        DataStream<VideoOrderDO> ds = env.addSource(new VideoOrderSource());

        WindowedStream<VideoOrderDO, String, TimeWindow> stream = ds.keyBy(new KeySelector<VideoOrderDO, String>() {
            @Override
            public String getKey(VideoOrderDO value) throws Exception {
                return value.getTitle();
            }
        }).window(TumblingProcessingTimeWindows.of(Time.seconds(5)));

        stream.apply(new WindowFunction<VideoOrderDO, Map<String,Object>, String, TimeWindow>() {
            @Override
            public void apply(String key, TimeWindow timeWindow, Iterable<VideoOrderDO> iterable, Collector<Map<String, Object>> collector) throws Exception {
                List<VideoOrderDO> list = IterableUtils.toStream(iterable).collect(Collectors.toList());
                long sum = list.stream().collect(Collectors.summarizingInt(VideoOrderDO::getMoney)).getSum();
                Map<String,Object> map = new HashMap<>();
                map.put("sumMoney",sum);
                map.put("title",key);
                collector.collect(map);
            }
        }).print();
        env.execute("apply Window job");
    }
}

process(new ProcessWindowFunction(){})

IN是输入类型，OUT是输出类型，KEY是分组类型，W是时间窗 
ProcessWindowFunction<IN, OUT, KEY, W extends Window>

/**
 * @author lixiang
 * process-Window滚动窗口
 */
public class FlinkWindow3Demo {
    public static void main(String[] args) throws Exception {
        //构建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        DataStream<VideoOrderDO> ds = env.addSource(new VideoOrderSource());

        WindowedStream<VideoOrderDO, String, TimeWindow> stream = ds.keyBy(new KeySelector<VideoOrderDO, String>() {
            @Override
            public String getKey(VideoOrderDO value) throws Exception {
                return value.getTitle();
            }
        }).window(TumblingProcessingTimeWindows.of(Time.seconds(5)));

        stream.process(new ProcessWindowFunction<VideoOrderDO, Map<String,Object>, String, TimeWindow>() {
            @Override
            public void process(String key, ProcessWindowFunction<VideoOrderDO, Map<String, Object>, String, TimeWindow>.Context context, Iterable<VideoOrderDO> iterable, Collector<Map<String, Object>> collector) throws Exception {
                List<VideoOrderDO> list = IterableUtils.toStream(iterable).collect(Collectors.toList());
                long sum = list.stream().collect(Collectors.summarizingInt(VideoOrderDO::getMoney)).getSum();
                Map<String,Object> map = new HashMap<>();
                map.put("sumMoney",sum);
                map.put("title",key);
                collector.collect(map);
            }
        }).print();

        env.execute("process Window job");
    }
}

aggregate(new AggregateFunction(){});

apply(new WindowFunction(){})

process(new ProcessWindowFunction(){}) //比WindowFunction功能强大

public class TimeUtil {

    /**
     * 时间处理
     * @param date
     * @return
     */
    public static String toDate(Date date){
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        ZoneId zoneId = ZoneId.systemDefault();
        return formatter.format(date.toInstant().atZone(zoneId));
    }

    /**
     * 字符串转日期类型
     * @param time
     * @return
     */
    public static Date strToDate(String time){
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        LocalDateTime dateTime = LocalDateTime.parse(time, formatter);
        return Date.from(dateTime.atZone(ZoneId.systemDefault()).toInstant());
    }

    /**
     * 时间处理
     * @param date
     * @return
     */
    public static String format(long date){
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        ZoneId zoneId = ZoneId.systemDefault();
        return formatter.format(new Date(date).toInstant().atZone(zoneId));
    }

}

/**
 * @author lixiang
 */
public class FlinkWaterDemo {
    public static void main(String[] args) throws Exception {

        //初始化环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度
        env.setParallelism(1);

        //监听socket输入
        DataStreamSource<String> source = env.socketTextStream("192.168.139.20", 8888);

        //一对多转换，将输入的字符串转成Tuple类型
        SingleOutputStreamOperator<Tuple3<String, String, Integer>> flatMap = source.flatMap(new FlatMapFunction<String, Tuple3<String, String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple3<String, String, Integer>> out) throws Exception {
                String[] split = value.split(",");
                out.collect(new Tuple3<String,String,Integer>(split[0],split[1],Integer.parseInt(split[2])));
            }
        });

        //设置watermark，官方文档直接拿来的，注意修改自己的时间参数
        SingleOutputStreamOperator<Tuple3<String, String, Integer>> watermarks = flatMap.assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple3<String, String, Integer>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner((element, recordTimestamp) -> {
                    return TimeUtil.strToDate(element.f1).getTime();
                }));

        //根据标题进行分组
        watermarks.keyBy(new KeySelector<Tuple3<String, String, Integer>, String>() {
            @Override
            public String getKey(Tuple3<String, String, Integer> value) throws Exception {
                return value.f0;
            }
        }).window(TumblingEventTimeWindows.of(Time.seconds(10))).apply(new WindowFunction<Tuple3<String, String, Integer>, String, String, TimeWindow>() {
            //滚动窗口，10s一统计，全窗口函数
            @Override
            public void apply(String key, TimeWindow timeWindow, Iterable<Tuple3<String, String, Integer>> iterable, Collector<String> collector) throws Exception {
                List<String> eventTimeList = new ArrayList<>();
                int total = 0;
                for (Tuple3<String, String, Integer> order : iterable) {
                    eventTimeList.add(order.f1);
                    total = total + order.f2;
                }
                String outStr = "分组key:"+key+",总价:"+total+",窗口开始时间:"+TimeUtil.format(timeWindow.getStart())+",窗口结束时间:"+TimeUtil.format(timeWindow.getEnd())+",窗口所有事件时间:"+eventTimeList;
                collector.collect(outStr);
            }
        }).print();

        env.execute("watermark job");
    }
}

[root@flink ~]# nc -lk 8888
java,2022-11-11 23:12:07,10
java,2022-11-11 23:12:11,10
java,2022-11-11 23:12:08,10
mysql,2022-11-11 23:12:13,20
java,2022-11-11 23:12:13,10
java,2022-11-11 23:12:17,10
java,2022-11-11 23:12:09,10
java,2022-11-11 23:12:20,10
java,2022-11-11 23:12:22,10
java,2022-11-11 23:12:25,10

/**
 * @author lixiang
 */
public class FlinkWaterDemo {
    public static void main(String[] args) throws Exception {

        //初始化环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度
        env.setParallelism(1);

        //监听socket输入
        DataStreamSource<String> source = env.socketTextStream("192.168.139.20", 8888);

        //一对多转换，将输入的字符串转成Tuple类型
        SingleOutputStreamOperator<Tuple3<String, String, Integer>> flatMap = source.flatMap(new FlatMapFunction<String, Tuple3<String, String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple3<String, String, Integer>> out) throws Exception {
                String[] split = value.split(",");
                out.collect(new Tuple3<String,String,Integer>(split[0],split[1],Integer.parseInt(split[2])));
            }
        });

        //设置watermark，官方文档直接拿来的，注意修改自己的时间参数
        SingleOutputStreamOperator<Tuple3<String, String, Integer>> watermarks = flatMap.assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple3<String, String, Integer>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner((element, recordTimestamp) -> {
                    return TimeUtil.strToDate(element.f1).getTime();
                }));

        //根据标题进行分组
        watermarks.keyBy(new KeySelector<Tuple3<String, String, Integer>, String>() {
            @Override
            public String getKey(Tuple3<String, String, Integer> value) throws Exception {
                return value.f0;
            }
        }).window(TumblingEventTimeWindows.of(Time.seconds(10)))
                //1min的容忍时间，即使时间段窗口被统计了，只要数据没有超过1min就可以再次被统计进去
                .allowedLateness(Time.minutes(1))
                .apply(new WindowFunction<Tuple3<String, String, Integer>, String, String, TimeWindow>() {
            //滚动窗口，10s一统计，全窗口函数
            @Override
            public void apply(String key, TimeWindow timeWindow, Iterable<Tuple3<String, String, Integer>> iterable, Collector<String> collector) throws Exception {
                List<String> eventTimeList = new ArrayList<>();
                int total = 0;
                for (Tuple3<String, String, Integer> order : iterable) {
                    eventTimeList.add(order.f1);
                    total = total + order.f2;
                }
                String outStr = "分组key:"+key+",总价:"+total+",窗口开始时间:"+TimeUtil.format(timeWindow.getStart())+",窗口结束时间:"+TimeUtil.format(timeWindow.getEnd())+",窗口所有事件时间:"+eventTimeList;
                collector.collect(outStr);
            }
        }).print();

        env.execute("watermark job");
    }
}

/**
 * @author lixiang
 */
public class FlinkWaterDemo {
    public static void main(String[] args) throws Exception {

        //初始化环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度
        env.setParallelism(1);

        //监听socket输入
        DataStreamSource<String> source = env.socketTextStream("192.168.139.20", 8888);

        //一对多转换，将输入的字符串转成Tuple类型
        SingleOutputStreamOperator<Tuple3<String, String, Integer>> flatMap = source.flatMap(new FlatMapFunction<String, Tuple3<String, String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple3<String, String, Integer>> out) throws Exception {
                String[] split = value.split(",");
                out.collect(new Tuple3<String,String,Integer>(split[0],split[1],Integer.parseInt(split[2])));
            }
        });

        //设置watermark，官方文档直接拿来的，注意修改自己的时间参数
        SingleOutputStreamOperator<Tuple3<String, String, Integer>> watermarks = flatMap.assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple3<String, String, Integer>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner((element, recordTimestamp) -> {
                    return TimeUtil.strToDate(element.f1).getTime();
                }));

        //new 一个OutputTag Bean
        OutputTag<Tuple3<String,String,Integer>> lateData = new OutputTag<Tuple3<String,String,Integer>>("lateData"){};

        //根据标题进行分组
        SingleOutputStreamOperator<String> operator = watermarks.keyBy(new KeySelector<Tuple3<String, String, Integer>, String>() {
                    @Override
                    public String getKey(Tuple3<String, String, Integer> value) throws Exception {
                        return value.f0;
                    }
                }).window(TumblingEventTimeWindows.of(Time.seconds(10)))
                //1min的容忍时间，即使时间段窗口被统计了，只要数据没有超过1min就可以再次被统计进去
                .allowedLateness(Time.minutes(1))
                //侧输入，最后的兜底数据
                .sideOutputLateData(lateData)
                .apply(new WindowFunction<Tuple3<String, String, Integer>, String, String, TimeWindow>() {
                    //滚动窗口，10s一统计，全窗口函数
                    @Override
                    public void apply(String key, TimeWindow timeWindow, Iterable<Tuple3<String, String, Integer>> iterable, Collector<String> collector) throws Exception {
                        List<String> eventTimeList = new ArrayList<>();
                        int total = 0;
                        for (Tuple3<String, String, Integer> order : iterable) {
                            eventTimeList.add(order.f1);
                            total = total + order.f2;
                        }
                        String outStr = "分组key:" + key + ",总价:" + total + ",窗口开始时间:" + TimeUtil.format(timeWindow.getStart()) + ",窗口结束时间:" + TimeUtil.format(timeWindow.getEnd()) + ",窗口所有事件时间:" + eventTimeList;
                        collector.collect(outStr);
                    }
                });

        operator.print();

        //侧输出流数据
        operator.getSideOutput(lateData).print();

        env.execute("watermark job");
    }
}

[root@flink ~]# nc -lk 8888
java,2022-11-11 23:12:07,10
java,2022-11-11 23:12:11,10
java,2022-11-11 23:12:08,10
mysql,2022-11-11 23:12:13,20
java,2022-11-11 23:12:13,10
java,2022-11-11 23:12:17,10
java,2022-11-11 23:12:09,10
java,2022-11-11 23:12:20,10
java,2022-11-11 23:14:22,10 
java,2022-11-11 23:12:25,10 #设置一个超过1分钟的数据，测试

新接口，`WatermarkStrategy`，`TimestampAssigner` 和 `WatermarkGenerator` 因为其对时间戳和 watermark 等重点的抽象和分离很清晰，并且还统一了周期性和标记形式的 watermark 生成方式

新接口之前是用AssignerWithPeriodicWatermarks和AssignerWithPunctuatedWatermarks ,现在可以弃用了

MemoryStateBackend（内存，不推荐在生产场景使用）
FsStateBackend（文件系统上，本地文件系统、HDFS, 性能更好,常用）
RocksDBStateBackend (无需担心 OOM 风险，是大部分时候的选择)

配置条目的可能值是hashmap (HashMapStateBackend)、rocksdb (EmbeddedRocksDBStateBackend) 
或实现状态后端工厂StateBackendFactory的类的完全限定类名

#全局配置例子一
# The backend that will be used to store operator state checkpoints
state.backend: hashmap

# Optional, Flink will automatically default to JobManagerCheckpointStorage
# when no checkpoint directory is specified.
state.checkpoint-storage: jobmanager

#全局配置例子二
state.backend: rocksdb
state.checkpoints.dir: file:///checkpoint-dir/

# Optional, Flink will automatically default to FileSystemCheckpointStorage
# when a checkpoint directory is specified.
state.checkpoint-storage: filesystem

//代码配置一
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(new HashMapStateBackend());
env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage());

//代码配置二
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(new EmbeddedRocksDBStateBackend());
env.getCheckpointConfig().setCheckpointStorage("file:///checkpoint-dir");
//或者
env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("file:///checkpoint-dir"));

<dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-statebackend-rocksdb_${scala.version}</artifactId>
            <version>1.13.1</version>
</dependency>

/**
 * 使用valueState实现maxBy功能，统计分组内订单金额最高的订单
 * @author lixiang
 */
public class FlinkStateDemo {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStream<String> ds = env.socketTextStream("192.168.139.20", 8888);

        DataStream<Tuple3<String, String, Integer>> flatMapDS = ds.flatMap(new RichFlatMapFunction<String, Tuple3<String, String, Integer>>() {

            @Override
            public void flatMap(String value, Collector<Tuple3<String, String, Integer>> out) throws Exception {
                String[] arr = value.split(",");
                out.collect(Tuple3.of(arr[0], arr[1], Integer.parseInt(arr[2])));
            }
        });

        //一定要key by后才可以使用键控状态ValueState
        SingleOutputStreamOperator<Tuple2<String, Integer>> maxVideoOrder = flatMapDS.keyBy(new KeySelector<Tuple3<String,String,Integer>, String>() {
            @Override
            public String getKey(Tuple3<String, String, Integer> value) throws Exception {
                return value.f0;
            }
        }).map(new RichMapFunction<Tuple3<String, String, Integer>, Tuple2<String, Integer>>() {
            private ValueState<Integer> valueState = null;
            @Override
            public void open(Configuration parameters) throws Exception {
                valueState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("total", Integer.class));
            }
            @Override
            public Tuple2<String, Integer> map(Tuple3<String, String, Integer> tuple3) throws Exception {
                // 取出State中的最大值
                Integer stateMaxValue = valueState.value();

                Integer currentValue = tuple3.f2;

                if (stateMaxValue == null || currentValue > stateMaxValue) {
                    //更新状态,把当前的作为新的最大值存到状态中
                    valueState.update(currentValue);
                    return Tuple2.of(tuple3.f0, currentValue);
                } else {
                    //历史值更大
                    return Tuple2.of(tuple3.f0, stateMaxValue);
                }
            }
        });
        maxVideoOrder.print();
        env.execute("valueState job");
    }
}

[root@flink ~]# nc -lk 8888
java,2022-11-11 23:12:07,10
java,2022-11-11 23:12:11,10
java,2022-11-11 23:12:08,30
mysql,2022-11-11 23:12:13,20
java,2022-11-11 23:12:13,10

//全局配置checkpoints
state.checkpoints.dir: hdfs:///checkpoints/

//作业单独配置checkpoints
env.getCheckpointConfig().setCheckpointStorage("hdfs:///checkpoints-data/");
//全局配置savepoint
state.savepoints.dir: hdfs:///flink/savepoints

数据一致性保证都是由流处理器实现的，也就是说都是在Flink流处理器内部保证的

在真实应用中，了流处理器以外还包含了数据源(例如Kafka、Mysql)和输出到持久化系统（Kafka、Mysql、Hbase、CK）

端到端的一致性保证，是意味着结果的正确性贯穿了整个流处理应用的各个环节，每一个组件都要保证自己的一致性。

//两个检查点之间间隔时间，默认是0,单位毫秒
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
//Checkpoint过程中出现错误，是否让整体任务都失败，默认值为0，表示不容忍任何Checkpoint失败
env.getCheckpointConfig().setTolerableCheckpointFailureNumber(5);

//Checkpoint是进行失败恢复，当一个 Flink 应用程序失败终止、人为取消等时，它的 Checkpoint 就会被清除
//可以配置不同策略进行操作
// DELETE_ON_CANCELLATION: 当作业取消时，Checkpoint 状态信息会被删除，因此取消任务后，不能从 Checkpoint 位置进行恢复任务
// RETAIN_ON_CANCELLATION(多): 当作业手动取消时，将会保留作业的 Checkpoint 状态信息,要手动清除该作业的 Checkpoint 状态信息
       env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

//Flink 默认提供 Extractly-Once 保证 State 的一致性，还提供了 Extractly-Once，At-Least-Once 两种模式，
// 设置checkpoint的模式为EXACTLY_ONCE，也是默认的，
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

//设置checkpoint的超时时间, 如果规定时间没完成则放弃，默认是10分钟
env.getCheckpointConfig().setCheckpointTimeout(60000);

//设置同一时刻有多少个checkpoint可以同时执行，默认为1就行，以避免占用太多正常数据处理资源
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

//设置了重启策略, 作业在失败后能自动恢复,失败后最多重启3次，每次重启间隔10s
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10000));

<dependency>
     <groupId>org.apache.flink</groupId>
     <artifactId>flink-cep-scala_2.11</artifactId>
     <version>1.7.0</version>
</dependency>

/**
 * cep-demo
 * @author lixiang
 */
public class FlinkCEPDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStream<String> ds = env.socketTextStream("192.168.139.20",8888);
        SingleOutputStreamOperator<Tuple3<String, String, Integer>> flatMapDS = ds.flatMap(new FlatMapFunction<String, Tuple3<String, String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple3<String, String, Integer>> out) throws Exception {
                String[] arr = value.split(",");
                out.collect(Tuple3.of(arr[0], arr[1], Integer.parseInt(arr[2])));
            }
        });
        SingleOutputStreamOperator<Tuple3<String, String, Integer>> watermarks = flatMapDS.assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple3<String, String, Integer>>forMonotonousTimestamps()
                .withTimestampAssigner((event, timestamp) -> TimeUtil.strToDate(event.f1).getTime()));

        KeyedStream<Tuple3<String, String, Integer>, String> keyedStream = watermarks.keyBy(new KeySelector<Tuple3<String, String, Integer>, String>() {
            @Override
            public String getKey(Tuple3<String, String, Integer> value) throws Exception {
                return value.f0;
            }
        });

        //定义模式
        Pattern<Tuple3<String, String, Integer>, Tuple3<String, String, Integer>> pattern = Pattern.<Tuple3<String, String, Integer>>
                        begin("firstTimeLogin")
                .where(new SimpleCondition<Tuple3<String, String, Integer>>() {
                    @Override
                    public boolean filter(Tuple3<String, String, Integer> value) throws Exception {
                        return value.f2 == -1;
                    }
                })
                .next("secondTimeLogin")
                .where(new SimpleCondition<Tuple3<String, String, Integer>>() {
                    @Override
                    public boolean filter(Tuple3<String, String, Integer> value) throws Exception {
                        return value.f2 == -1;
                    }
                }).within(Time.seconds(5));

        //匹配检查
        PatternStream<Tuple3<String, String, Integer>> patternStream = CEP.pattern(keyedStream, pattern);
        SingleOutputStreamOperator<Tuple3<String, String, String>> select = patternStream.select(new PatternSelectFunction<Tuple3<String, String, Integer>, Tuple3<String, String, String>>() {
            @Override
            public Tuple3<String, String, String> select(Map<String, List<Tuple3<String, String, Integer>>> map) throws Exception {
                Tuple3<String, String, Integer> firstLoginFail = map.get("firstTimeLogin").get(0);
                Tuple3<String, String, Integer> secondLoginFail = map.get("secondTimeLogin").get(0);
                return Tuple3.of(firstLoginFail.f0, firstLoginFail.f1, secondLoginFail.f1);
            }
        });

        select.print("匹配结果");

        env.execute("CEP job");
    }
}

张三,2022-11-11 12:01:01,-1
李四,2022-11-11 12:01:10,-1
李四,2022-11-11 12:01:11,-1
张三,2022-11-11 12:01:13,-1
李四,2022-11-11 12:01:14,-1
李四,2022-11-11 12:01:15,1
张三,2022-11-11 12:01:16,-1
李四,2022-11-11 12:01:17,-1
张三,2022-11-11 12:01:20,1

（1）上传jdk1.8安装包，解压到指目录
tar -xvf jdk-8u181-linux-x64.tar.gz -C /usr/local/

（2）查看解压后的文件
[root@flink ~]# cd /usr/local
[root@flink local]# ls
bin  etc  flink-1.13.1  games  include  jdk1.8.0_181  lib  lib64  libexec  sbin  share  src

（3）jdk1.8.0_181重命名为jdk1.8
mv jdk1.8.0_181/ jdk1.8

（4）配置环境变量
vi /etc/profile
添加配置：
JAVA_HOME=/usr/local/jdk1.8
CLASSPATH=$JAVA_HOME/lib/
PATH=$PATH:$JAVA_HOME/bin
export PATH JAVA_HOME CLASSPATH

（5）刷新配置
source /etc/profile

（6）查看java环境
[root@flink ~]# java -version
java version "1.8.0_181"
Java(TM) SE Runtime Environment (build 1.8.0_181-b13)
Java HotSpot(TM) 64-Bit Server VM (build 25.181-b13, mixed mode)

Please specify JAVA_HOME. Either in Flink config ./conf/flink-conf.yaml or as system-wide JAVA_HOME.

tar -xvf flink-1.13.1-bin-scala_2.12.tgz -C /usr/local/

调整配置文件：conf/flink-conf.yaml

#web ui 端口
rest.port=8081

#调整jobmanager和taskmanager的大小，根据自己的机器进行调整
jobmanager.memory.process.size: 256m
taskmanager.memory.process.size: 256m

本地模式用到这两个脚本
start-cluster.sh
stop-cluster.sh

启动本地模式：./start-cluster.sh
注意这会可能会报错：
The derived from fraction jvm overhead memory (19.200mb (20132659 bytes)) is less than its min value 192.000mb (201326592 bytes), min value will be used instead

原因分析：
flink最少需要192M的内存才能启动，jdk1.8之后堆内存采用的是元空间，元空间是根据内存大小自动分配的。如果元空间给的太小，flink将会启动不起来。

调整如下配置：conf/flink-conf.yaml
taskmanager.memory.process.size: 512m
 
taskmanager.memory.framework.heap.size: 64m
taskmanager.memory.framework.off-heap.size: 64m
taskmanager.memory.jvm-metaspace.size: 64m
taskmanager.memory.jvm-overhead.fraction: 0.2
taskmanager.memory.jvm-overhead.min: 16m
taskmanager.memory.jvm-overhead.max: 64m
 
taskmanager.memory.network.fraction: 0.1
taskmanager.memory.network.min: 1mb
taskmanager.memory.network.max: 256mb