Flink-FastSteps

概念：

一、应用场景：

1、Data Pipeline
实时数仓
2、Data Analytics
实时大屏
实时报表
3、Data Driven
风控系统

二、windows&watermark

1
2
3
4
5
6
7
8
9
10
//开窗测试 指定窗口分配器
DataStream<Integer> resultStream = dataStream.keyBy("id")
        //设置一个15秒的一个滚动窗口
        .window(TumblingProcessingTimeWindows.of(Time.seconds(15)));
		//滑动窗口
        //.window(SlidingProcessingTimeWindows.of(Time.seconds(20),Time.seconds(10)))
        //会话窗口
        //.window(ProcessingTimeSessionWindows.withGap(Time.seconds(15)))

env.execute();

  • Tumbling Windows：统计每一分钟中用户购买的商品的总数，需要将用户的行为事件按每一分钟进行切分，这种切分被成为翻滚时间窗口；
  • Slidding Windows：每30秒计算一次最近一分钟用户购买的商品总数；
  • Session Windows：一计算每个用户在活跃期间总共购买的商品数量，如果用户30秒没有活动则视为会话断开

Window解决了什么问题？
在流处理应用中，数据是连续不断的，因此我们不可能等到所有数据都到了才开始处理。当然我们可以每来一个消息就处理一次，但是有时我们需要做一些聚合类的处理，例如：在过去的1分钟内有多少用户点击了我们的网页。在这种情况下，我们必须定义一个窗口，用来收集最近一分钟内的数据，并对这个窗口内的数据进行计算。

Watermark解决了什么问题？
实时系统中，由于各种原因造成的延时，造成某些消息发到flink的时间延时于事件产生的时间。如果基于event time构建window，但是对于late element，我们又不能无限期的等下去，必须要有个机制来保证一个特定的时间后，必须触发window去进行计算了。这个特别的机制，就是watermark。默认是丢弃，也可以通过配置allowedLateness()延迟丢弃

1
2
3
4
val outputWindow: DataStream[String] = waterMarkStream
      .keyBy(0)
      .window(TumblingEventTimeWindows.of(Time.seconds(3)))
      // .allowedLateness(Time.seconds(2))//允许数据迟到2S

剖析 Window API
Flink 中定义一个窗口主要需要以下三个组件。

  • Window Assigner：用来决定某个元素被分配到哪个/哪些窗口中去。如下类图展示了目前内置实现的 Window Assigners：
  • Trigger：触发器。决定了一个窗口何时能触发计算输出结果，每个窗口都会拥有一个自己的Trigger。
  • Evictor：可以译为“驱逐者”。在Trigger触发之后，在窗口被处理之前，Evictor（如果有Evictor的话）会用来剔除窗口中不需要的元素，相当于一个filter。

上述三个组件的不同实现的不同组合，可以定义出非常复杂的窗口。Flink 中内置的窗口也都是基于这三个组件构成的，当然内置窗口有时候无法解决用户特殊的需求，所以 Flink 也暴露了这些窗口机制的内部接口供用户实现自定义的窗口。

窗口在类别上有两大类：
时间窗口.timewindow()和计数窗口.countwindow()

其中时间窗口有两种使用方法：
一个是.timewindow()通过输入的参数区分窗口类别。

1
2
3
4
5
6
7
8
9
10
11
12
//滚动窗口
.timeWindow(Time.seconds(15))
//滑动窗口
.timeWindow(Time.seconds(15),Time.seconds(5))
//会话窗口
.window(ProcessingTimeSessionWindows.withGap(Time.seconds(15)))

//计数滚动窗口
.countWindow(10)
//滑动计数窗口
.countWindow(10,2)

一个是.window()在参数中声明窗口

1
2
3
4
5
6
7
8
9
10
11
//滚动窗口
.window(TumblingProcessingTimeWindows.of(Time.seconds(15)));
//滑动窗口
.window(SlidingProcessingTimeWindows.of(Time.seconds(20),Time.seconds(10)))
//会话窗口
.window(ProcessingTimeSessionWindows.withGap(Time.seconds(15)))

//计数滚动窗口
.countWindow(10)
//滑动计数窗口
.countWindow(10,2)

窗口函数分为两类：增量聚合函数、全量聚合函数
增量聚合函数：ReduceFunction、AggregateFunction

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
//对窗口进行聚合操作 增量窗口操作
.timeWindow(Time.seconds(15))
.aggregate(new AggregateFunction<SensorReading, Integer, Integer>() {
    @Override
    //创建累加器
    public Integer createAccumulator() {
        return 0;
    }
    @Override
    public Integer add(SensorReading sensorReading, Integer accumulator) {
        return accumulator+1;
    }
    @Override
    public Integer getResult(Integer accumulator) {
        return accumulator;
    }
    @Override
    public Integer merge(Integer integer, Integer acc1) {
        return null;
    }
});

全量聚合函数：ProcessWindowFunction，WindowFunction

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
.timeWindow(Time.seconds(15))
.apply(new WindowFunction<SensorReading, Tuple3<String,Long,Integer>,
Tuple, TimeWindow>() {

    @Override
    public void apply(Tuple tuple, 
                      TimeWindow window, 
                      Iterable<SensorReading> input, 
                      Collector<Tuple3<String,Long,Integer>> out) throws Exception {
        String id =tuple.getField(0);
        Long windowEnd =window.getEnd();
        Integer count = IteratorUtils.toList(input.iterator()).size();
        out.collect(new Tuple3<>(id,windowEnd,count));
    }
});