day04_Flink高级API

Posted ChinaManor

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了day04_Flink高级API相关的知识，希望对你有一定的参考价值。

day04_Flink高级API

今日目标

  • Flink的四大基石
  • Flink窗口Window操作
  • Flink时间 - Time
  • Flink水印 - Watermark机制
  • Flink的state状态管理-keyed state 和 operator state

Flink的四大基石

  • Checkpoint - 检查点， 分布式一致性，解决数据丢失，故障恢复数据， 存储的是全局的状态， 持久化HDFS分布式文件系统中
  • State - 状态，分为Managed state（托管状态） 和 Rawed state （原始状态）； 数据结构的角度来说 ValueState、ListState、MapState，BroadcastState
  • Time - 时间 ， EventTime事件时间、Ingestion摄取时间、Process处理时间
  • Window - 窗口，时间窗口 和 计数窗口， TimeWindow 、 countwindow、 sessionwindow

Window操作

  • 为什么需要 Window - 窗口

    数据是动态的， 无界的， 需要窗口划定范围，将无界数据转换成有界、静态的数据进行计算。

Window分类

  • time - 时间进行分类

    • 时间的窗口级别， 一天，一小时，一分钟

    • 用的比较多 滚动窗口 - tumbling window 和 滑动窗口 - sliding window

    • 滚动窗口 ，窗口时间和滑动时间一样就是滚动时间

    • 滑动窗口， 滑动的时间小于窗口的时间；

    • 会话窗口 - session windows

  • count - 计数进行分类

    • 滚动计数窗口
    • 滑动计数窗口

如何使用

image-20210507090957187

windows的案例

时间窗口需求

  • 每5秒钟统计一次，最近5秒钟内，各个路口通过红绿灯汽车的数量–基于时间的滚动窗口
  • 每5秒钟统计一次，最近10秒钟内，各个路口通过红绿灯汽车的数量–基于时间的滑动窗口
package cn.itcast.flink.basestone;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.MapFunction;
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.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * Author itcast
 * Date 2021/6/18 15:00
 * 开发步骤
 * 1. 将 字符串 9,3 转换成 CartInfo
 * 2. 使用 滚动窗口， 滑动窗口
 * 3. 分组和聚合
 * 4. 打印输出
 * 5. 执行环境
 */
public class WindowDemo01 {
    public static void main(String[] args) throws Exception {
        //1.env 创建流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //2.读取 socket 数据源
        DataStreamSource<String> source = env.socketTextStream("192.168.88.161", 9999);
        //3.将9,3转为CartInfo(9,3)
        DataStream<CartInfo> mapDS = source.map(new MapFunction<String, CartInfo>() {
            @Override
            public CartInfo map(String value) throws Exception {
                String[] kv = value.split(",");
                return new CartInfo(kv[0], Integer.parseInt(kv[1]));
            }
        });
        //4.按照 sensorId 分组并划分滚动窗口为5秒，在窗口上求和
        // Tumbling(滚动)Processing(处理)TimeWindows(时间窗口)
        //需求1:每5秒钟统计一次，最近5秒钟内，各个路口/信号灯通过红绿灯汽车的数量
        SingleOutputStreamOperator<CartInfo> result1 = mapDS.keyBy(t -> t.sensorId)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .sum("count");
        //需求2:每5秒钟统计一次，最近10秒钟内，各个路口/信号灯通过红绿灯汽车的数量
        SingleOutputStreamOperator<CartInfo> result2 = mapDS.keyBy(t -> t.sensorId)
                .window(SlidingProcessingTimeWindows.of(Time.seconds(10),Time.seconds(5)))
                .sum("count");
        //5.打印输出
        //result1.print();
        result2.print();
        //6.execute
        env.execute();
    }

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class CartInfo {
        private String sensorId;//信号灯id
        private Integer count;//通过该信号灯的车的数量
    }
}

计数窗口需求

  • 需求1:统计在最近5条消息中,各自路口通过的汽车数量,相同的key每出现5次进行统计–基于数量的滚动窗口

  • 需求2:统计在最近5条消息中,各自路口通过的汽车数量,相同的key每出现3次进行统计–基于数量的滑动窗口

package cn.itcast.flink.basestone;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.MapFunction;
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;

/**
 * Author itcast
 * Date 2021/6/18 15:46
 * Desc TODO
 */
public class CountWindowDemo01 {
    public static void main(String[] args) throws Exception {
        //1.env 创建流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //2.读取 socket 数据源
        DataStreamSource<String> source = env.socketTextStream("192.168.88.161", 9999);
        //3.将9,3转为CartInfo(9,3)
        DataStream<WindowDemo01.CartInfo> mapDS = source.map(new MapFunction<String, WindowDemo01.CartInfo>() {
            @Override
            public WindowDemo01.CartInfo map(String value) throws Exception {
                String[] kv = value.split(",");
                return new WindowDemo01.CartInfo(kv[0], Integer.parseInt(kv[1]));
            }
        });
        // * 需求1:统计在最近5条消息中,各自路口通过的汽车数量,相同的key每出现5次进行统计--基于数量的滚动窗口
        //        //countWindow(long size, long slide)
        SingleOutputStreamOperator<WindowDemo01.CartInfo> result1 = mapDS.keyBy(t -> t.getSensorId())
                .countWindow(5)
                .sum("count");
        // * 需求2:统计在最近5条消息中,各自路口通过的汽车数量,相同的key每出现3次进行统计--基于数量的滑动窗口
        SingleOutputStreamOperator<WindowDemo01.CartInfo> result2 = mapDS.keyBy(t -> t.getSensorId())
                .countWindow(5, 3)
                .sum("count");

        //打印输出
        //result1.print();
        result2.print();
        //执行环境
        env.execute();
    }
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class CartInfo {
        private String sensorId;//信号灯id
        private Integer count;//通过该信号灯的车的数量
    }
}
package cn.itcast.flink.basestone;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.MapFunction;
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;

/**
 * Author itcast
 * Date 2021/6/18 15:46
 * Desc TODO
 */
public class CountWindowDemo01 {
    public static void main(String[] args) throws Exception {
        //1.env 创建流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //2.读取 socket 数据源
        DataStreamSource<String> source = env.socketTextStream("192.168.88.161", 9999);
        //3.将9,3转为CartInfo(9,3)
        DataStream<WindowDemo01.CartInfo> mapDS = source.map(new MapFunction<String, WindowDemo01.CartInfo>() {
            @Override
            public WindowDemo01.CartInfo map(String value) throws Exception {
                String[] kv = value.split(",");
                return new WindowDemo01.CartInfo(kv[0], Integer.parseInt(kv[1]));
            }
        });
        // * 需求1:统计在最近5条消息中,各自路口通过的汽车数量,相同的key每出现5次进行统计--基于数量的滚动窗口
        //        //countWindow(long size, long slide)
        SingleOutputStreamOperator<WindowDemo01.CartInfo> result1 = mapDS.keyBy(t -> t.getSensorId())
                .countWindow(5)
                .sum("count");
        // * 需求2:统计在最近5条消息中,各自路口通过的汽车数量,相同的key每出现3次进行统计--基于数量的滑动窗口
        SingleOutputStreamOperator<WindowDemo01.CartInfo> result2 = mapDS.keyBy(t -> t.getSensorId())
                .countWindow(5, 3)
                .sum("count");

        //打印输出
        //result1.print();
        result2.print();
        //执行环境
        env.execute();
    }
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class CartInfo {
        private String sensorId;//信号灯id
        private Integer count;//通过该信号灯的车的数量
    }
}

Flink - Time 和 watermark

Time - 时间

image-20210618160007265

水印机制 - watermark

  • 主要解决数据延迟问题

  • 水印（时间戳） = 事件时间 - 允许最大的延时时间

  • 窗口触发条件

    水印时间 >= 窗口的结束时间 触发计算

image-20210618161943842

需求

有订单数据,格式为: (订单ID，用户ID，时间戳/事件时间，订单金额)

要求每隔5s, 计算5秒内，每个用户的订单总金额

并添加Watermark来解决一定程度上的数据延迟和数据乱序（最多延时 3 秒）问题。

package cn.itcast.flink.basestone;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;
import java.util.Random;
import java.util.UUID;

/**
 * Author itcast
 * Date 2021/6/18 16:54
 * Desc TODO
 */
public class WatermarkDemo01 {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置属性 ProcessingTime  ， 新版本 默认设置 EventTime
        //env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        //2.Source 创建 Order 类 orderId:String userId:Integer money:Integer eventTime:Long
        DataStreamSource<Order> source = env.addSource(new SourceFunction<Order>() {
            boolean flag = true;
            Random rm = new Random();

            @Override
            public void run(SourceContext<Order> ctx) throws Exception {
                while (flag) {
                    ctx.collect(new Order(
                            UUID.randomUUID().toString(),
                            rm.nextInt(3),
                            rm.nextInt(101),
                            //模拟生成 Order 数据  事件时间=当前时间-5秒钟随机*1000
                            System.currentTimeMillis() - rm.nextInt(5) * 1000
                    ));
                    Thread.sleep(1000);
                }
            }

            @Override
            public void cancel() {
                flag = false;
            }
        });

        //3.Transformation
        //-告诉Flink要基于事件时间来计算!
        //env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);//新版本默认就是EventTime
        DataStream<Order> result = source.assignTimestampsAndWatermarks(
                WatermarkStrategy.<Order>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((element, recordTimestamp) -> element.eventTime)
        )
                //-分配水印机制，最多延迟3秒，告诉Flink数据中的哪一列是事件时间,因为Watermark = 当前最大的事件时间 - 最大允许的延迟时间或乱序时间
                //代码走到这里,就已经被添加上Watermark了!接下来就可以进行窗口计算了
                //要求每隔5s,计算5秒内(基于时间的滚动窗口)，每个用户的订单总金额
                .keyBy(t -> t.userId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sum("money");
        //4.Sink
        result.print();
        //5.execute
        env.execute();
    }
    //创建订单类
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class Order{
        private String orderId;
        private Integer userId;
        private Integer money;
        private Long eventTime;
    }
}
  • 自定义重写接口实现水印机制
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;
import java.util.Random;
import java.util.UUID;

/**
 * Author itcast
 * Date 2021/6/18 16:54
 * Desc TODO
 */
public class WatermarkDemo01 {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置属性 ProcessingTime  ， 新版本 默认设置 EventTime
        //env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        //2.Source 创建 Order 类 orderId:String userId:Integer money:Integer eventTime:Long
        DataStreamSource<Order> source = env.addSource(new SourceFunction<Order>() {
            boolean flag = true;
            Random rm = new Random();

            @Override
            public void run(SourceContext<Order> ctx) throws Exception {
                while (flag) {
                    ctx.collect(new Order(
                            UUID.randomUUID().toString(),
                            rm.nextInt(3),
                            rm.nextInt(101),
                            //模拟生成 Order 数据  事件时间=当前时间-5秒钟随机*1000
                            System.currentTimeMillis() - rm.nextInt(5) * 1000
                    ));
                    Thread.sleep以上是关于day04_Flink高级API的主要内容，如果未能解决你的问题，请参考以下文章
 

 .~day08_高级特性和新特性


 20.Flink高级特性--新特性--双流Joinjoin的分类API代码演示-WindowJoin代码演示-IntervalJoin


 20.Flink高级特性--新特性--双流Joinjoin的分类API代码演示-WindowJoin代码演示-IntervalJoin


 flink学习day03：flink datastream 开发


 2021年最新最全Flink系列教程_Flink流批一体API


 2021年最新最全Flink系列教程_Flink原理初探和流批一体API