42讲轻松通关Flink学习笔记（中）

灰太狼 2022-12-06 15:50 339阅读 0赞

### 文章目录 ###

*  第十五讲 如何排查生产环境中的反压问题
 *  第十六讲 如何处理生产环境中的数据倾斜问题
 *  第十七讲 生产环境的并行度和资源设置
 *  第十八讲 如何进行生产环境作业监控
 *  第十九讲 Flink如何做维表关联
 *  第二十讲 Flink高级应用之海量数据高效去重
 *  第二十一讲 Flink 在实时计算平台和实时数据仓库中的作用
 *  第二十二讲 项目背景和整体架构设计
 *  第二十三讲 Mock Kafka 消息并发送
 *  第二十四讲 Flink 消费 Kafka 数据业务开发
 *  第二十五讲 Flink 中 watermark 的定义和使用
 *  第二十六讲 Flink中的聚合函数和累加器的设计和使用
 *  第二十七讲 Flink Redis Sink实现
 *  第二十八讲 TopN 热门商品功能实现

# 第十五讲 如何排查生产环境中的反压问题 #

这一课时我们主要讲解生产环境中 Flink 任务经常会遇到的一个问题，即如何处理好反压问题将直接关系到任务的资源使用和稳定运行。

反压问题是流式计算系统中经常碰到的一个问题，如果你的任务出现反压节点，那么就意味着任务数据的消费速度小于数据的生产速度，需要对生产数据的速度进行控制。通常情况下，反压经常出现在促销、热门活动等场景，它们有一个共同的特点：短时间内流量陡增造成数据的堆积或者消费速度变慢。

**不同框架的反压对比**

目前主流的大数据实时处理系统都对反压问题进行了专门的处理，希望框架自身能检测到被阻塞的算子，然后降低数据生产者的发送速率。我们所熟悉的 Storm、Spark Streaming、Flink 的实现稍微有所不同。

**Storm**

Storm 从 1.0 版本以后引入了全新的反压机制，Storm 会主动监控工作节点。当工作节点接收数据超过一定的水位值时，那么反压信息会被发送到 ZooKeeper 上，然后 ZooKeeper 通知所有的工作节点进入反压状态，最后数据的生产源头会降低数据的发送速度。

**Spark Streaming**

Spark Streaming 在原有的架构基础上专门设计了一个 RateController 组件，该组件利用经典的 PID 算法。向系统反馈当前系统处理数据的几个重要属性：消息数量、调度时间、处理时间、调度时间等，然后根据这些参数计算出一个速率，该速率则是当前系统处理数据的最大能力，Spark Streaming 会根据计算结果对生产者进行限速。

**Flink**

Flink 的反压设计利用了网络传输和动态限流。在 Flink 的设计哲学中，纯流式计算给 Flink 进行反压设计提供了天然的优势。

我们在以前的课程中讲解过，Flink 任务的组成由基本的“流”和“算子”构成，那么“流”中的数据在“算子”间进行计算和转换时，会被放入分布式的阻塞队列中。当消费者的阻塞队列满时，则会降低生产者的数据生产速度。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center]  
如上图所示，我们看一下 Flink 进行逐级反压的过程。当 Task C 的数据处理速度发生异常时，Receive Buffer 会呈现出队列满的情况，Task B 的 Send Buffer 会感知到这一点，然后把数据发送速度降低。以此类推，整个反压会一直从下向上传递到 Source 端；反之，当下游的 Task 处理能力有提升后，会在此反馈到 Source Task，数据的发送和读取速率都会升高，提高了整个任务的处理能力。

**反压的定位**

当你的任务出现反压时，如果你的上游是类似 Kafka 的消息系统，很明显的表现就是消费速度变慢，Kafka 消息出现堆积。

如果你的业务对数据延迟要求并不高，那么反压其实并没有很大的影响。但是对于规模很大的集群中的大作业，反压会造成严重的“并发症”。首先任务状态会变得很大，因为数据大规模堆积在系统中，这些暂时不被处理的数据同样会被放到“状态”中。另外，Flink 会因为数据堆积和处理速度变慢导致 checkpoint 超时，而 checkpoint 是 Flink 保证数据一致性的关键所在，最终会导致数据的不一致发生。

那么我们应该如何发现任务是否出现反压了呢？

**Flink Web UI**

Flink 的后台页面是我们发现反压问题的第一选择。Flink 的后台页面可以直观、清晰地看到当前作业的运行状态。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 1]  
如上图所示，是 Flink 官网给出的计算反压状态的案例。需要注意的是，只有用户在访问点击某一个作业时，才会触发反压状态的计算。在默认的设置下，Flink 的 TaskManager 会每隔 50 ms 触发一次反压状态监测，共监测 100 次，并将计算结果反馈给 JobManager，最后由 JobManager 进行计算反压的比例，然后进行展示。

这个比例展示逻辑如下：

*  OK: 0 <= Ratio <= 0.10，正常；
 *  LOW: 0.10 < Ratio <= 0.5，一般；
 *  HIGH: 0.5 < Ratio <= 1，严重。

官网同样给出了不同反压状态下，Flink Web UI 中任务运行的状态，如下图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 2]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 3]  
**Flink Metrics**

如果你想对 Flink 做更为详细的监控的话，Flink 本身提供了大量的 REST API 来获取任务的各种状态。[Flink 提供的所有系统监控指标你都点击这里找到][Flink]。

随着版本的持续变更，截止 1.10.0 版本，Flink 提供的监控指标中与反压最为密切的如下表所示：

<table> 
 <thead> 
  <tr> 
   <th>指标名称</th> 
   <th>用途</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>outPoolUsage</td> 
   <td>发送端缓冲池的使用率</td> 
  </tr> 
  <tr> 
   <td>inPoolUsage</td> 
   <td>接收端缓冲池的使用率</td> 
  </tr> 
  <tr> 
   <td>floatingBuffersUsage</td> 
   <td>处理节点缓冲池的使用率</td> 
  </tr> 
  <tr> 
   <td>exclusiveBuffersUsage</td> 
   <td>数据输入方缓冲池的使用率</td> 
  </tr> 
 </tbody> 
</table>

我们逐个介绍一下这四个指标。

*  **outPoolUsage**  
    这个指标代表的是当前 Task 的数据发送速率，当一个 Task 的 outPoolUsage 很高，则代表着数据发送速度很快。但是当一个 Task 的 outPoolUsage 很低，那么就需要特别注意，有可能是下游的处理速度很低导致的，也有可能当前节点就是反压节点，导致数据处理速度很慢。
 *  **inPoolUsage**  
    inPoolUsage 表示当前 Task 的数据接收速率，通常会和 outPoolUsage 配合使用；如果一个节点的 inPoolUsage 很高而 outPoolUsage 很低，则这个节点很有可能就是反压节点。
 *  **floatingBuffersUsage 和 exclusiveBuffersUsage**  
    floatingBuffersUsage 表示处理节点缓冲池的使用率；exclusiveBuffersUsage 表示数据输入通道缓冲池的使用率。

**反压问题处理**

我们已经知道反压产生的原因和监控的方法，当线上任务出现反压时，需要如何处理呢？

主要通过以下几个方面进行定位和处理：

*  数据倾斜
 *  GC
 *  代码本身

**数据倾斜**

数据倾斜问题是我们生产环境中出现频率最多的影响任务运行的因素，可以在 Flink 的后台管理页面看到每个 Task 处理数据的大小。当数据倾斜出现时，通常是简单地使用类似 KeyBy 等分组聚合函数导致的，需要用户将热点 Key 进行预处理，降低或者消除热点 Key 的影响。

**GC**

垃圾回收问题也是造成反压的因素之一。不合理的设置 TaskManager 的垃圾回收参数会导致严重的 GC 问题，我们可以通过 -XX:+PrintGCDetails 参数查看 GC 的日志。

**代码本身**

开发者错误地使用 Flink 算子，没有深入了解算子的实现机制导致性能问题。我们可以通过查看运行机器节点的 CPU 和内存情况定位问题。

# 第十六讲 如何处理生产环境中的数据倾斜问题 #

无论是对于 Flink、Spark 这样的实时计算框架还是 Hive 等离线计算框架，数据量从来都不是问题，真正引起问题导致严重后果的是数据倾斜。所谓数据倾斜，是指在大规模并行处理的数据中，其中某个运行节点处理的数据远远超过其他部分，这会导致该节点压力极大，最终出现运行失败从而导致整个任务的失败。

我们在这一课时中将分析出现数据倾斜的原因，Flink 任务中最容易出现数据倾斜的几个算子并且给出解决方案。

**数据倾斜背景和危害**

数据倾斜产生的原因和危害和解决方案有哪些呢？我们一一来看。

**数据倾斜原理**

目前我们所知道的大数据处理框架，比如 Flink、Spark、Hadoop 等之所以能处理高达千亿的数据，是因为这些框架都利用了分布式计算的思想，集群中多个计算节点并行，使得数据处理能力能得到线性扩展。

我们在第 03 课时“Flink 的编程模型与其他框架比较”中曾经讲过，在实际生产中 Flink 都是以集群的形式在运行，在运行的过程中包含了两类进程。其中 TaskManager 实际负责执行计算的 Worker，在其上执行 Flink Job 的一组 Task，Task 则是我们执行具体代码逻辑的容器。理论上只要我们的任务 Task 足够多就可以对足够大的数据量进行处理。

但是实际上大数据量经常出现，一个 Flink 作业包含 200 个 Task 节点，其中有 199 个节点可以在很短的时间内完成计算。但是有一个节点执行时间远超其他结果，并且随着数据量的持续增加，导致该计算节点挂掉，从而整个任务失败重启。我们可以在 Flink 的管理界面中看到任务的某一个 Task 数据量远超其他节点。

**数据倾斜原因和解决方案**

Flink 任务出现数据倾斜的直观表现是任务节点频繁出现反压，但是增加并行度后并不能解决问题；部分节点出现 OOM 异常，是因为大量的数据集中在某个节点上，导致该节点内存被爆，任务失败重启。

产生数据倾斜的原因主要有 2 个方面：

*  业务上有严重的数据热点，比如滴滴打车的订单数据中北京、上海等几个城市的订单量远远超过其他地区；
 *  技术上大量使用了 KeyBy、GroupBy 等操作，错误的使用了分组 Key，人为产生数据热点。

因此解决问题的思路也很清晰：

*  业务上要尽量避免热点 key 的设计，例如我们可以把北京、上海等热点城市分成不同的区域，并进行单独处理；
 *  技术上出现热点时，要调整方案打散原来的 key，避免直接聚合；此外 Flink 还提供了大量的功能可以避免数据倾斜。

那么我们就从典型的场景入手，看看在 Flink 任务中出现数据倾斜的主要场景和解决方案。

**Flink 任务数据倾斜场景和解决方案**

**两阶段聚合解决 KeyBy 热点**

KeyBy 是我们经常使用的分组聚合函数之一。在实际的业务中经常会碰到这样的场景：双十一按照下单用户所在的省聚合求订单量最高的前 10 个省，或者按照用户的手机类型聚合求访问量最高的设备类型等。

上述场景在我们进行 KeyBy 时就会出现严重的数据倾斜，如下图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 4]  
如果我们直接简单地使用 KeyBy 算子，模拟一个简单的统计 PV 的场景如下：

DataStream sourceStream = ...;
    windowedStream = sourceStream.keyBy("type")
              .window(TumblingEventTimeWindows.of(Time.minutes(1)));
    windowedStream.process(new MyPVFunction())
                  .addSink(new MySink())... 
    env.execute()...

我们在根据 type 进行 KeyBy 时，如果数据的 type 分布不均匀就会导致大量的数据分配到一个 task 中去，发生数据倾斜。

那么我们的解决思路是：

*  首先把分组的 key 打散，比如加随机后缀；
 *  对打散后的数据进行聚合；
 *  把打散的 key 还原为真正的 key；
 *  二次 KeyBy 进行结果统计，然后输出。

DataStream sourceStream = ...;
    resultStream = sourceStream
         .map(record -> { 
            Record record = JSON.parseObject(record, Record.class);
            String type = record.getType();
            record.setType(type + "#" + new Random().nextInt(100));
            return record;
          })
          .keyBy(0)
          .window(TumblingEventTimeWindows.of(Time.minutes(1)))
          .aggregate(new CountAggregate())
          .map(count -> { 
            String key = count.getKey.substring(0, count.getKey.indexOf("#"));
            return RecordCount(key,count.getCount);
          })
          //二次聚合
          .keyBy(0)
          .process(new CountProcessFunction);
    
    resultStream.sink()...
    env.execute()...

其中 CountAggregate 函数实现如下：

public class CountAggregate implements AggregateFunction<Record,CountRecord,CountRecord> { 
        @Override
        public CountRecord createAccumulator() { 
            return new CountRecord(null, 0L);
        }
        @Override
        public CountRecord add(Record value, CountRecord accumulator) { 
            if(accumulator.getKey() == null){ 
                accumulator.setKey(value.key);
            }
            accumulator.setCount(value.count);
            return accumulator;
        }
        @Override
        public CountRecord getResult(CountRecord accumulator) { 
            return accumulator;
        }
        @Override
        public CountRecord merge(CountRecord a, CountRecord b) { 
            return new CountRecord(a.getKey(),a.getCount()+b.getCount()) ;
        }
    }

CountProcessFunction 的实现如下：

public class CountProcessFunction extends KeyedProcessFunction<String, CountRecord, CountRecord> { 
        private ValueState<Long> state = this.getRuntimeContext().getState(new ValueStateDescriptor("count",Long.class));
        @Override
        public void processElement(CountRecord value, Context ctx, Collector<CountRecord> out) throws Exception { 
            if(state.value()==0){ 
                state.update(value.count);
                ctx.timerService().registerProcessingTimeTimer(ctx.timerService().currentProcessingTime() + 1000L * 5);
            }else{ 
                state.update(state.value() + value.count);
            }
        }
        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<CountRecord> out) throws Exception { 
            //这里可以做业务操作，例如每 5 分钟将统计结果发送出去
            //out.collect(...);
            //清除状态
            state.clear();
            //其他操作
            ...
        }
    }

通过上面打散聚合再二次聚合的方式，我们就可以实现热点 Key 的打散，消除数据倾斜。

**GroupBy + Aggregation 分组聚合热点问题**

业务上通过 GroupBy 进行分组，然后紧跟一个 SUM、COUNT 等聚合操作是非常常见的。我们都知道 GroupBy 函数会根据 Key 进行分组，完全依赖 Key 的设计，如果 Key 出现热点，那么会导致巨大的 shuffle，相同 key 的数据会被发往同一个处理节点；如果某个 key 的数据量过大则会直接导致该节点成为计算瓶颈，引起反压。

我们还是按照上面的分组统计 PV 的场景，SQL 语句如下：

select
          date,
          type,
          sum(count) as pv
    from table
          group by
          date,
          type;

我们可以通过内外两层聚合的方式将 SQL 改写为：

select date,
           type,
           sum(pv) as pv
    from(
      select
            date,
            type,
            sum(count) as pv
      from table
            group by
            date,
            type,
            floor(rand()*100) --随机打散成100份 
        )
        group by 
        date,
        type;

在上面的 SQL 拆成了内外两层，第一层通过随机打散 100 份的方式减少数据热点，当然这个打散的方式可以根据业务灵活指定。

**Flink 消费 Kafka 上下游并行度不一致导致的数据倾斜**

通常我们在使用 Flink 处理实时业务时，上游一般都是消息系统，Kafka 是使用最广泛的大数据消息系统。当使用 Flink 消费 Kafka 数据时，也会出现数据倾斜。

需要十分注意的是，我们 Flink 消费 Kafka 的数据时，是推荐上下游并行度保持一致，即 Kafka 的分区数等于 Flink Consumer 的并行度。

但是会有一种情况，为了加快数据的处理速度，来设置 Flink 消费者的并行度大于 Kafka 的分区数。如果你不做任何的设置则会导致部分 Flink Consumer 线程永远消费不到数据。

这时候你需要设置 Flink 的 Redistributing，也就是数据重分配。

Flink 提供了多达 8 种重分区策略，类图如下图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 5]  
在我们接收到 Kafka 消息后，可以通过自定义数据分区策略来实现数据的负载均衡，例如：

dataStream
            .setParallelism(2)
            // 采用REBALANCE分区策略重分区
            .rebalance() //.rescale()
            .print()
            .setParallelism(4);

其中，Rebalance 分区策略，数据会以 round-robin 的方式对数据进行再次分区，可以全局负载均衡。

Rescale 分区策略基于上下游的并行度，会将数据以循环的方式输出到下游的每个实例中。

**其他**

Flink 一直在不断地迭代，不断出现各种各样的手段解决我们遇到的数据倾斜问题。例如，MiniBatch 微批处理手段等，需要我们开发者不断地去发现，并学习新的解决问题的办法。

# 第十七讲 生产环境的并行度和资源设置 #

在使用 Flink 处理生产实际问题时，并行度和资源的配置调优是我们经常要面对的工作之一，如何有效和正确地配置并行度是我们的任务能够高效执行的必要条件。这一课时就来看一下生产环境的并行度和资源配置问题。

**Flink 中的计算资源**

通常我们说的 Flink 中的计算资源是指具体任务的 Task。首先要理解 Flink 中的计算资源的一些核心概念，比如 Slot、Chain、Task 等，正确理解这些概念有助于开发者了解 Flink 中的计算资源是如何进行隔离和管理的，也有助于我们快速地定位生产中的问题。

**Task Slot**

我们在第 03 课时“Flink 的编程模型与其他框架比较” 中提到过，在实际生产中，Flink 都是以集群在运行，在运行的过程中包含了两类进程，其中之一就是：TaskManager。

在 Flink 集群中，一个 TaskManger 就是一个 JVM 进程，并且会用独立的线程来执行 task，为了控制一个 TaskManger 能接受多少个 task，Flink 提出了 Task Slot 的概念。

我们可以简单地把 Task Slot 理解为 TaskManager 的计算资源子集。假如一个 TaskManager 拥有 5 个 Slot，那么该 TaskManager 的计算资源会被平均分为 5 份，不同的 task 在不同的 Slot 中执行，避免资源竞争。但需要注意的是，Slot 仅仅用来做内存的隔离，对 CPU 不起作用。那么运行在同一个 JVM 的 task 可以共享 TCP 连接，以减少网络传输，在一定程度上提高了程序的运行效率，降低了资源消耗。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 6]  
**Slot 共享**

默认情况下，Flink 还允许同一个 Job 的子任务共享 Slot。因为在一个 Flink 任务中，有很多的算子，这些算子的计算压力各不相同，比如简单的 map 和 filter 算子所需要的资源不多，但是有些算子比如 window、group by 则需要更多的计算资源才能满足计算所需。这时候那些资源需求大的算子就可以共用其他的 Slot，提高整个集群的资源利用率。

**Operator Chain**

此外，Flink 自身会把不同的算子的 task 连接在一起组成一个新的 task。这么做是因为 Flink 本身提供了非常有效的任务优化手段，因为 task 是在同一个线程中执行，那么可以有效减少线程间上下文的切换，并且减少序列化/反序列化带来的资源消耗，从而在整体上提高我们任务的吞吐量。

**并行度**

Flink 使用并行度来定义某一个算子被切分成多少个子任务。我们的 Flink 代码会被转换成逻辑视图，在实际运行时，根据用户的并行度配置会被转换成对应的子任务进行执行。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 7]  
**源码解析**

Flink Job 在执行中会通过 SlotProvider 向 ResourceManager 申请资源，RM 负责协调 TaskManager，满足 JobManager 的资源请求。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 8]  
整体的类图如上图所示，SlotProvider 中的 allocateSlot 方法负责向 SlotPool 申请可用的 slot 资源，通过 returnLogicSlot 将空闲的 slot 释放至 SlotPool。

在整个 Flink 的资源管理的类中，核心的类包括 Scheduler、SlotPool、JobMaster。它们之间的交互流程主要包括：Scheduler 调度器会向 SlotPool 资源池申请和释放 Slot；如果 SlotPool 不能满足需求，那么会向 ResourceManager 发起申请；获取到的资源通过 JobMaster 分配给 SlotPool。

关于更多的资源管理的实现流程，可以参考上面的类图查看源码。

**如何设置并行度**

Flink 本身支持不同级别来设置我们任务并行度的方法，它们分别是：

*  算子级别
 *  环境级别
 *  客户端级别
 *  集群配置级别

下面依次讲解 Flink 中的四种并行度的设置方法，以及它们的优先级。

**算子级别**

我们在编写 Flink 程序时，可以在代码中显示的制定不同算子的并行度。用经典的 wordcount 程序举例：

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
     
    DataStream<String> text = ...
     
    DataStream<Tuple2<String, Integer>> wordCounts = text
    .flatMap(new LineSplitter())
    .setParallelism(10)
    .keyBy(0)
    .timeWindow(Time.seconds(5))
    .sum(1).setParallelism(1);
    wordCounts.print();
    env.execute("word count");

如上述代码所示，我们可以通过显示的调用 setParallelism() 方法来显示的指定每个算子的并行度配置。

在实际生产中，我们推荐在算子级别显示指定各自的并行度，方便进行显示和精确的资源控制。

**环境级别**

环境级别的并行度设置指的是我们可以通过调用 env.setParallelism() 方法来设置整个任务的并行度：

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(5);
    ...

一旦设置了这个参数，表明我们任务中的所有算子的并行度都是指定的值，生产环境中并不推荐。

**客户端级别**

我们可以在使用命令提交 Flink Job 的时候指定并行度，当任务执行时发现代码中没有设置并行度，便会采用我们提交命令时的参数。

通过 -p 命令来指定任务提交时候的并行度：

./bin/flink run -p 5 ../wordCount-java*.jar

**集群配置级别**

我们的 flink-conf.yaml 文件中有一个参数 parallelism.default，该参数会在用户不设置任何其他的并行度配置时生效：  
![在这里插入图片描述][20200917110704465.png_pic_center]  
需要特别指出的是，设置并行度的优先级依次是：

**算子级别 > 环境级别 > 客户端级别 > 集群配置级别**

我们推荐在生产环境中使用算子级别的并行度进行资源控制。

# 第十八讲 如何进行生产环境作业监控 #

本课时主要讲解如何进行生产环境作业监控。

在第 15 课时“如何排查生产环境中的反压问题”中提到过我们应该如何发现任务是否出现反压，Flink 的后台页面是我们发现反压问题的第一选择，其后台页面可以直观、清晰地看到当前作业的运行状态。

在实际生产中，Flink 的后台页面可以方便我们对 Flink JobManager、TaskManager、执行计划、Slot 分配、是否反压等参数进行定位，对单个任务来讲可以方便地进行问题排查。

但是，对于很多大中型企业来讲，我们对进群的作业进行管理时，更多的是关心作业精细化实时运行状态。例如，实时吞吐量的同比环比、整个集群的任务运行概览、集群水位，或者监控利用 Flink 实现的 ETL 框架的运行情况等，这时候就需要设计专门的监控系统来监控集群的任务作业情况。

**Flink Metrics**

针对上面的情况，我们就用了 Flink 提供的另一个强大的功能：Flink Metrics。

Flink Metrics 是 Flink 实现的一套运行信息收集库，我们不但可以收集 Flink 本身提供的系统指标，比如 CPU、内存、线程使用情况、JVM 垃圾收集情况、网络和 IO 等，还可以通过继承和实现指定的类或者接口打点收集用户自定义的指标。

通过使用 Flink Metrics 我们可以轻松地做到：

*  实时采集 Flink 中的 Metrics 信息或者自定义用户需要的指标信息并进行展示；
 *  通过 Flink 提供的 Rest API 收集这些信息，并且接入第三方系统进行展示。

**Flink Metrics 分类**

Flink 提供了四种类型的监控指标，分别是：Counter、Gauge、Histogram、Meter。

**Counter**

Counter 称为计数器，一般用来统计其中一个指标的总量，比如统计数据的输入、输出总量。

public class MyMapper extends RichMapFunction<String, String> { 
      private transient Counter counter;
      @Override
      public void open(Configuration config) { 
        this.counter = getRuntimeContext()
          .getMetricGroup()
          .counter("MyCounter");
      }
      @Override
      public String map(String value) throws Exception { 
        this.counter.inc();
        return value;
      }
    }

**Gauge**

Gauge 被用来统计某一个指标的瞬时值。举个例子，我们在监控 Flink 中某一个节点的内存使用情况或者某个 map 算子的输出值数量。

public class MyMapper extends RichMapFunction<String, String> { 
      private transient int valueNumber = 0L;
      @Override
      public void open(Configuration config) { 
        getRuntimeContext()
          .getMetricGroup()
          .gauge("MyGauge", new Gauge<Long>() { 
            @Override
            public Long getValue() { 
              return valueNumber;
            }
          });
      }
      @Override
      public String map(String value) throws Exception { 
        valueNumber++;
        return value;
      }
    }

**Meter**

Meter 被用来计算一个指标的平均值。

public class MyMapper extends RichMapFunction<Long, Integer> { 
      private Meter meter;
      @Override
      public void open(Configuration config) { 
        this.meter = getRuntimeContext()
          .getMetricGroup()
          .meter("myMeter", new MyMeter());
      }
      @public Integer map(Long value) throws Exception { 
        this.meter.markEvent();
      }
    }

**Histogram**

Histogram 是直方图，Flink 中属于直方图的指标非常少，它通常被用来计算指标的最大值、最小值、中位数等。

public class MyMapper extends RichMapFunction<Long, Integer> { 
      private Histogram histogram;
      @Override
      public void open(Configuration config) { 
        this.histogram = getRuntimeContext()
          .getMetricGroup()
          .histogram("myHistogram", new MyHistogram());
      }
      @public Integer map(Long value) throws Exception { 
        this.histogram.update(value);
      }
    }

这里需要特别指出的是，Flink 中的 Metrics 是一个多层的结构，以 Group 的方式存在，我们用来定位唯一的一个 Metrics 是通过 Metric Group + Metric Name 的方式。

**源码分析**

Flink Metrics 相关的实现都是通过 org.apache.flink.metrics.Metric 这个类实现的，整体的类图如下所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 9]  
为了方便对 Metrics 进行管理和分类，Flink 提供了对 Metrics 进行分组的功能，这个功能是通过下图中 MetricGroup 实现的，在图中可以看到 MetricGroup 相关的子类的继承关系。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 10]  
此外，Flink 还提供了向外披露 Metric 的监测结果的接口，该接口是 org.apache.flink.metrics.reporter.MetricReporter。这个接口的实现类通过 Metrics 的类型进行注册和移除。

public abstract class AbstractReporter implements MetricReporter, CharacterFilter { 
        protected final Logger log = LoggerFactory.getLogger(this.getClass());
        protected final Map<Gauge<?>, String> gauges = new HashMap();
        protected final Map<Counter, String> counters = new HashMap();
        protected final Map<Histogram, String> histograms = new HashMap();
        protected final Map<Meter, String> meters = new HashMap();
        public AbstractReporter() { 
        }
        public void notifyOfAddedMetric(Metric metric, String metricName, MetricGroup group) { 
            String name = group.getMetricIdentifier(metricName, this);
            synchronized(this) { 
                if(metric instanceof Counter) { 
                    this.counters.put((Counter)metric, name);
                } else if(metric instanceof Gauge) { 
                    this.gauges.put((Gauge)metric, name);
                } else if(metric instanceof Histogram) { 
                    this.histograms.put((Histogram)metric, name);
                } else if(metric instanceof Meter) { 
                    this.meters.put((Meter)metric, name);
                } else { 
                    this.log.warn("Cannot add unknown metric type {}. This indicates that the reporter does not support this metric type.", metric.getClass().getName());
                }
            }
        }
        public void notifyOfRemovedMetric(Metric metric, String metricName, MetricGroup group) { 
            synchronized(this) { 
                if(metric instanceof Counter) { 
                    this.counters.remove(metric);
                } else if(metric instanceof Gauge) { 
                    this.gauges.remove(metric);
                } else if(metric instanceof Histogram) { 
                    this.histograms.remove(metric);
                } else if(metric instanceof Meter) { 
                    this.meters.remove(metric);
                } else { 
                    this.log.warn("Cannot remove unknown metric type {}. This indicates that the reporter does not support this metric type.", metric.getClass().getName());
                }
            }
        }
    }

**获取 Metrics**

获取 Metrics 的方法有多种，首先我们可以通过 Flink 的后台管理页面看到部分指标；其次可以通过 Flink 提供的 Http 接口查询 Flink 任务的状态信息，因为 Flink Http 接口返回的都是 Json 信息，我们可以很方便地将 Json 进行解析；最后一种方法是，我们可以通过 Metric Reporter 获取。下面分别对这两者进行详细讲解。

**Flink HTTP 接口**

Flink 提供了丰富的接口来协助我们查询 Flink 中任务运行的状态，所有的请求都可以通过访问 `http://hostname:8081/` 加指定的 URI 方式查询，Flink 支持的所有 HTTP 接口你都可以点击这里查询到。

Flink 支持的接口包括：

/config
    /overview
    /jobs
    /joboverview/running
    /joboverview/completed
    /jobs/<jobid>
    /jobs/<jobid>/vertices
    /jobs/<jobid>/config
    /jobs/<jobid>/exceptions
    /jobs/<jobid>/accumulators
    /jobs/<jobid>/vertices/<vertexid>
    /jobs/<jobid>/vertices/<vertexid>/subtasktimes
    /jobs/<jobid>/vertices/<vertexid>/taskmanagers
    /jobs/<jobid>/vertices/<vertexid>/accumulators
    /jobs/<jobid>/vertices/<vertexid>/subtasks/accumulators
    /jobs/<jobid>/vertices/<vertexid>/subtasks/<subtasknum>
    /jobs/<jobid>/vertices/<vertexid>/subtasks/<subtasknum>/attempts/<attempt>
    /jobs/<jobid>/vertices/<vertexid>/subtasks/<subtasknum>/attempts/<attempt>/accumulators
    /jobs/<jobid>/plan
    /jars/upload
    /jars
    /jars/:jarid
    /jars/:jarid/plan
    /jars/:jarid/run

举个例子，我们可以通过查询 /joboverview 访问集群中所有任务的概览，结果类似如下形式：

{ 
      "running":[],
      "finished":[
        { 
          "jid": "7684be6004e4e955c2a558a9bc463f65",
          "name": "Flink Java Job at Wed Sep 16 18:08:21 CEST 2015",
          "state": "FINISHED",
          "start-time": 1442419702857,
          "end-time": 1442419975312,
          "duration":272455,
          "last-modification": 1442419975312,
          "tasks": { 
             "total": 6,
             "pending": 0,
             "running": 0,
             "finished": 6,
             "canceling": 0,
             "canceled": 0,
             "failed": 0
          }
        },
        { 
          "jid": "49306f94d0920216b636e8dd503a6409",
          "name": "Flink Java Job at Wed Sep 16 18:16:39 CEST 2015",
          ...
        }]
    }

**Flink Reporter**

Flink 还提供了很多内置的 Reporter，这些 Reporter 在 Flink 的[官网][Link 1]中可以查询到。

例如，Flink 提供了 Graphite、InfluxDB、Prometheus 等内置的 Reporter，我们可以方便地对这些外部系统进行集成。关于它们的详细配置也可以在 Flink 官网的详情页面中看到。

这里我们举一个 Flink 和 InfluxDB、Grafana 集成进行 Flink 集群任务监控的案例。在这个监控系统中，InfluxDB 扮演了 Flink 中监控数据存储者的角色，Grafana 则扮演了数据展示者的角色。

**InfluxDB 的安装**

> InfluxDB 是一个由 InfluxData 开发的开源时序型数据，由 Go 写成，着力于高性能地查询与存储时序型数据。InfluxDB 被广泛应用于存储系统的监控数据，IoT 行业的实时数据等场景。

InfluxDB 的安装过程很简单，我们不在这里赘述了，需要注意的事项是修改 InfluxDB 的配置 /etc/influxdb/influxdb.conf：

[admin]
       enabled = true
       bind-address = ":8083"

我们就可以通过 8083 端口打开 InfluxDB 的控制台了。

**Grafana 的安装**

[安装可以直接点击这里参考官网说明][Link 2]，Grafana 的默认账号和密码分别是 admin、admin，可以通过 3000 端口进行访问。

**修改 flink-conf.yaml**

我们需要在 flink 的配置文件中新增以下配置：

metrics.reporter.influxdb.class: org.apache.flink.metrics.influxdb.InfluxdbReporter
    metrics.reporter.influxdb.host: xxx.xxx.xxx.xxx
    metrics.reporter.influxdb.port: 8086
    metrics.reporter.influxdb.db: flink

同时，将 flink-metrics-influxdb-1.10.0.jar 这个包复制到 flink 的 /lib 目录下，然后启动 Flink。  
我们就可以在 Grafana 中看到 Metrics 信息了。

事实上，常用的 Flink 实时监控大盘包括但不限于：Prometheus+Grafana、Flink 日志接入 ELK等可以供用户选择。结合易用性、稳定性和接入成本，综合考虑，我们推荐实际监控中可以采用 Prometheus/InfluxDB+Grafana 相配合的方式。

# 第十九讲 Flink如何做维表关联 #

在实际生产中，我们经常会有这样的需求，需要以原始数据流作为基础，然后关联大量的外部表来补充一些属性。例如，我们在订单数据中，希望能得到订单收货人所在省的名称，一般来说订单中会记录一个省的 ID，那么需要根据 ID 去查询外部的维度表补充省名称属性。

在 Flink 流式计算中，我们的一些维度属性一般存储在 MySQL/HBase/Redis 中，这些维表数据存在定时更新，需要我们根据业务进行关联。根据我们业务对维表数据关联的时效性要求，有以下几种解决方案：

*  实时查询维表
 *  预加载全量数据
 *  LRU 缓存
 *  其他

上述几种关联外部维表的方式几乎涵盖了我们所有的业务场景，下面针对这几种关联维表的方式和特点一一讲解它们的实现方式和注意事项。

**实时查询维表**

实时查询维表是指用户在 Flink 算子中直接访问外部数据库，比如用 MySQL 来进行关联，这种方式是同步方式，数据保证是最新的。但是，当我们的流计算数据过大，会对外部系统带来巨大的访问压力，一旦出现比如连接失败、线程池满等情况，由于我们是同步调用，所以一般会导致线程阻塞、Task 等待数据返回，影响整体任务的吞吐量。而且这种方案对外部系统的 QPS 要求较高，在大数据实时计算场景下，QPS 远远高于普通的后台系统，峰值高达十万到几十万，整体作业瓶颈转移到外部系统。

这种方式的核心是，我们可以在 Flink 的 Map 算子中建立访问外部系统的连接。下面以订单数据为例，我们根据下单用户的城市 ID，去关联城市名称，核心代码实现如下：

public class Order { 
        private Integer cityId;
        private String userName;
        private String items;
        public Integer getCityId() { 
            return cityId;
        }
        public void setCityId(Integer cityId) { 
            this.cityId = cityId;
        }
        public String getUserName() { 
            return userName;
        }
        public void setUserName(String userName) { 
            this.userName = userName;
        }
        public String getItems() { 
            return items;
        }
        public void setItems(String items) { 
            this.items = items;
        }
        @Override
        public String toString() { 
            return "Order{" +
                    "cityId=" + cityId +
                    ", userName='" + userName + '\'' +
                    ", items='" + items + '\'' +
                    '}';
        }
    }
    public class DimSync extends RichMapFunction<String,Order> { 
        private static final Logger LOGGER = LoggerFactory.getLogger(DimSync.class);
        private Connection conn = null;
        public void open(Configuration parameters) throws Exception { 
            super.open(parameters);
            conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/dim?characterEncoding=UTF-8", "admin", "admin");
        }
        public Order map(String in) throws Exception { 
            JSONObject jsonObject = JSONObject.parseObject(in);
            Integer cityId = jsonObject.getInteger("city_id");
            String userName = jsonObject.getString("user_name");
            String items = jsonObject.getString("items");
            //根据city_id 查询 city_name
            PreparedStatement pst = conn.prepareStatement("select city_name from info where city_id = ?");
            pst.setInt(1,cityId);
            ResultSet resultSet = pst.executeQuery();
            String cityName = null;
            while (resultSet.next()){ 
                cityName = resultSet.getString(1);
            }
            pst.close();
            return new Order(cityId,userName,items,cityName);
        }
        public void close() throws Exception { 
            super.close();
            conn.close();
        }
    }

在上面这段代码中，RichMapFunction 中封装了整个查询维表，然后进行关联这个过程。需要注意的是，一般我们在查询小数据量的维表情况下才使用这种方式，并且要妥善处理连接外部系统的线程，一般还会用到线程池。最后，为了保证连接及时关闭和释放，一定要在最后的 close 方式释放连接，否则会将 MySQL 的连接数打满导致任务失败。

**预加载全量数据**

全量预加载数据是为了解决每条数据流经我们的数据系统都会对外部系统发起访问，以及对外部系统频繁访问而导致的连接和性能问题。这种思路是，每当我们的系统启动时，就将维度表数据全部加载到内存中，然后数据在内存中进行关联，不需要直接访问外部数据库。

这种方式的优势是我们只需要一次性地访问外部数据库，大大提高了效率。但问题在于，一旦我们的维表数据发生更新，那么 Flink 任务是无法感知的，可能会出现维表数据不一致，针对这种情况我们可以采取定时拉取维表数据。并且这种方式由于是将维表数据缓存在内存中，对计算节点的内存消耗很高，所以不能适用于数量很大的维度表。

我们还是用上面的场景，根据下单用户的城市 ID 去关联城市名称，核心代码实现如下：

public class WholeLoad extends RichMapFunction<String,Order> { 
        private static final Logger LOGGER = LoggerFactory.getLogger(WholeLoad.class);
        ScheduledExecutorService executor = null;
        private Map<String,String> cache;
        @Override
        public void open(Configuration parameters) throws Exception { 
            super.open(parameters);
            executor.scheduleAtFixedRate(new Runnable() { 
                @Override
                public void run() { 
                    try { 
                        load();
                    } catch (Exception e) { 
                        e.printStackTrace();
                    }
                }
            },5,5, TimeUnit.MINUTES);
        }
        @Override
        public Order map(String value) throws Exception { 
            JSONObject jsonObject = JSONObject.parseObject(value);
            Integer cityId = jsonObject.getInteger("city_id");
            String userName = jsonObject.getString("user_name");
            String items = jsonObject.getString("items");
            String cityName = cache.get(cityId);
            return new Order(cityId,userName,items,cityName);
        }
        public void load() throws Exception { 
            Class.forName("com.mysql.jdbc.Driver");
            Connection con = DriverManager.getConnection("jdbc:mysql://localhost:3306/dim?characterEncoding=UTF-8", "admin", "admin");
            PreparedStatement statement = con.prepareStatement("select city_id,city_name from info");
            ResultSet rs = statement.executeQuery();
            while (rs.next()) { 
                String cityId = rs.getString("city_id");
                String cityName = rs.getString("city_name");
                cache.put(cityId, cityName);
            }
            con.close();
        }
    }

在上面的例子中，我们使用 ScheduledExecutorService 每隔 5 分钟拉取一次维表数据。这种方式适用于那些实时场景不是很高，维表数据较小的场景。

**LRU 缓存**

LRU 是一种缓存算法，意思是最近最少使用的数据则被淘汰。在这种策略中，我们的维表数据天然的被分为冷数据和热数据。所谓冷数据指的是那些不经常使用的数据，热数据是那些查询频率高的数据。

对应到我们上面的场景中，根据城市 ID 关联城市的名称，北京、上海这些城市的订单远远高于偏远地区的一些城市，那么北京、上海就是热数据，偏远城市就是冷数据。这种方式存在一定的数据延迟，并且需要额外设置每条数据的失效时间。因为热点数据由于经常被使用，会常驻我们的缓存中，一旦维表发生变更是无法感知数据变化的。在这里使用 Guava 库提供的 CacheBuilder 来创建我们的缓存：

CacheBuilder.newBuilder()
            //最多存储10000条
            .maximumSize(10000)
            //过期时间为1分钟
            .expireAfterWrite(60, TimeUnit.SECONDS)
            .build();

整体的实现思路是：我们利用 Flink 的 RichAsyncFunction 读取 Hbase 的数据到缓存中，我们在关联维度表时先去查询缓存，如果缓存中不存在这条数据，就利用客户端去查询 Hbase，然后插入到缓存中。

首先我们需要一个 Hbase 的异步客户端：

<dependency>
        <groupId>org.hbase</groupId>
        <artifactId>asynchbase</artifactId>
        <version>1.8.2</version>
    </dependency>

核心的代码实现如下：

public class LRU extends RichAsyncFunction<String,Order> { 
        private static final Logger LOGGER = LoggerFactory.getLogger(LRU.class);
        String table = "info";
        Cache<String, String> cache = null;
        private HBaseClient client = null;
        @Override
        public void open(Configuration parameters) throws Exception { 
            super.open(parameters);
            //创建hbase客户端
            client = new HBaseClient("127.0.0.1","7071");
            cache = CacheBuilder.newBuilder()
                    //最多存储10000条
                    .maximumSize(10000)
                    //过期时间为1分钟
                    .expireAfterWrite(60, TimeUnit.SECONDS)
                    .build();
        }
        @Override
        public void asyncInvoke(String input, ResultFuture<Order> resultFuture) throws Exception { 
            JSONObject jsonObject = JSONObject.parseObject(input);
            Integer cityId = jsonObject.getInteger("city_id");
            String userName = jsonObject.getString("user_name");
            String items = jsonObject.getString("items");
            //读缓存
            String cacheCityName = cache.getIfPresent(cityId);
            //如果缓存获取失败再从hbase获取维度数据
            if(cacheCityName != null){ 
                Order order = new Order();
                order.setCityId(cityId);
                order.setItems(items);
                order.setUserName(userName);
                order.setCityName(cacheCityName);
                resultFuture.complete(Collections.singleton(order));
            }else { 
                client.get(new GetRequest(table,String.valueOf(cityId))).addCallback((Callback<String, ArrayList<KeyValue>>) arg -> { 
                    for (KeyValue kv : arg) { 
                        String value = new String(kv.value());
                        Order order = new Order();
                        order.setCityId(cityId);
                        order.setItems(items);
                        order.setUserName(userName);
                        order.setCityName(value);
                        resultFuture.complete(Collections.singleton(order));
                        cache.put(String.valueOf(cityId), value);
                    }
                    return null;
                });
            }
        }
    }

这里需要特别注意的是，我们用到了异步 IO (RichAsyncFunction)，这个功能的出现就是为了解决与外部系统交互时网络延迟成为系统瓶颈的问题。

我们在流计算环境中，在查询外部维表时，假如访问是同步进行的，那么整体能力势必受限于外部系统。正是因为异步 IO 的出现使得访问外部系统可以并发的进行，并且不需要同步等待返回，大大减轻了因为网络等待时间等引起的系统吞吐和延迟问题。

我们在使用异步 IO 时，一定要使用异步客户端，如果没有异步客户端我们可以自己创建线程池模拟异步请求。

**其他**

除了上述常见的处理方式，我们还可以通过将维表消息广播出去，或者自定义异步线程池访问维表，甚至还可以自己扩展 Flink SQL 中关联维表的方式直接使用 SQL Join 方法关联查询结果。

总体来讲，关联维表的方式就以上几种方式，并且基于这几种方式还会衍生出各种各样的解决方案。我们在评价一个方案的优劣时，应该从业务本身出发，不同的业务场景下使用不同的方式。

# 第二十讲 Flink高级应用之海量数据高效去重 #

消除重复数据是我们在实际业务中经常遇到的一类问题。在大数据领域，重复数据的删除有助于减少存储所需要的存储容量。而且在一些特定的业务场景中，重复数据是不可接受的，例如，精确统计网站一天的用户数量、在事实表中统计每天发出的快递包裹数量。在传统的离线计算中，我们可以直接用 SQL 通过 DISTINCT 函数，或者数据量继续增加时会用到类似 MapReduce 的思想。那么在实时计算中，去重计数是一个增量和长期的过程，并且不同的场景下因为效率和精度问题方案也需要变化。

针对上述问题，我们在这里列出几种常见的 Flink 中实时去重方案：

*  基于状态后端
 *  基于 HyperLogLog
 *  基于布隆过滤器（BloomFilter）
 *  基于 BitMap
 *  基于外部数据库

下面我们依次讲解上述几种方案的适用场景和实现原理。

**基于状态后端**

我们在第 09 课时“Flink 状态与容错”中曾经讲过 Flink State 状态后端的概念和原理，其中状态后端的种类之一是 RocksDBStateBackend。它会将正在运行中的状态数据保存在 RocksDB 数据库中，该数据库默认将数据存储在 TaskManager 运行节点的数据目录下。

RocksDB 是一个 K-V 数据库，我们可以利用 MapState 进行去重。这里我们模拟一个场景，计算每个商品 SKU 的访问量。

整体的实现代码如下：

public class MapStateDistinctFunction extends KeyedProcessFunction<String,Tuple2<String,Integer>,Tuple2<String,Integer>> { 
        private transient ValueState<Integer> counts;
        @Override
        public void open(Configuration parameters) throws Exception { 
            //我们设置 ValueState 的 TTL 的生命周期为24小时，到期自动清除状态
            StateTtlConfig ttlConfig = StateTtlConfig
                    .newBuilder(org.apache.flink.api.common.time.Time.minutes(24 * 60))
                    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                    .build();
            //设置 ValueState 的默认值
            ValueStateDescriptor<Integer> descriptor = new ValueStateDescriptor<Integer>("skuNum", Integer.class);
            descriptor.enableTimeToLive(ttlConfig);
            counts = getRuntimeContext().getState(descriptor);
            super.open(parameters);
        }
    
        @Override
        public void processElement(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception { 
            String f0 = value.f0;
            //如果不存在则新增
            if(counts.value() == null){ 
                counts.update(1);
            }else{ 
                //如果存在则加1
                counts.update(counts.value()+1);
            }
            
            out.collect(Tuple2.of(f0, counts.value()));
        }
    }

在上面的这段代码里，我们实现了这样的逻辑：定义了一个 MapStateDistinctFunction 类，该类继承了 KeyedProcessFunction。核心的处理逻辑在 processElement 方法中，当一条数据经过，我们会在 MapState 中判断这条数据是否已经存在，如果不存在那么计数为 1，如果存在，那么在原来的计数上加 1。

这里需要注意的是，我们自定义了状态的过期时间是 24 小时，在实际生产中大量的 Key 会使得状态膨胀，我们可以对存储的 Key 进行处理。例如，使用加密方法把 Key 加密成几个字节进再存储。

**基于 HyperLogLog**

> HyperLogLog 是一种估计统计算法，被用来统计一个集合中不同数据的个数，也就是我们所说的去重统计。HyperLogLog 算法是用于基数统计的算法，每个 HyperLogLog 键只需要花费 12 KB 内存，就可以计算接近 2 的 64 方个不同元素的基数。HyperLogLog 适用于大数据量的统计，因为成本相对来说是更低的，最多也就占用 12KB 内存。

我们在不需要 100% 精确的业务场景下，可以使用这种方法进行统计。首先新增依赖：

我们还是以上述的商品 SKU 访问量作为业务场景，数据格式为 <SKU, 访问的用户 id>:

public class HyperLogLogDistinct implements AggregateFunction<Tuple2<String,Long>,HLL,Long> { 
        @Override
        public HLL createAccumulator() { 
            return new HLL(14, 5);
        }
        @Override
        public HLL add(Tuple2<String, Long> value, HLL accumulator) { 
            //value 为访问记录 <商品sku, 用户id>
            accumulator.addRaw(value.f1);
            return accumulator;
        }
        @Override
        public Long getResult(HLL accumulator) { 
            long cardinality = accumulator.cardinality();
            return cardinality;
        }
        @Override
        public HLL merge(HLL a, HLL b) { 
            a.union(b);
            return a;
        }
    }

在上面的代码中，addRaw 方法用于向 HyperLogLog 中插入元素。如果插入的元素非数值型的，则需要 hash 过后才能插入。accumulator.cardinality() 方法用于计算 HyperLogLog 中元素的基数。

需要注意的是，HyperLogLog 并不是精准的去重，如果业务场景追求 100% 正确，那么一定不要使用这种方法。

**基于布隆过滤器（BloomFilter）**

> BloomFilter（布隆过滤器）类似于一个 HashSet，用于快速判断某个元素是否存在于集合中，其典型的应用场景就是能够快速判断一个 key 是否存在于某容器，不存在就直接返回。

需要注意的是，和 HyperLogLog 一样，布隆过滤器不能保证 100% 精确。但是它的插入和查询效率都很高。

我们可以在非精确统计的情况下使用这种方法：

public class BloomFilterDistinct extends KeyedProcessFunction<Long, String, Long> { 
        private transient ValueState<BloomFilter> bloomState;
        private transient ValueState<Long> countState;
    
        @Override
        public void processElement(String value, Context ctx, Collector<Long> out) throws Exception { 
            BloomFilter bloomFilter = bloomState.value();
            Long skuCount = countState.value();
            if(bloomFilter == null){ 
                BloomFilter.create(Funnels.unencodedCharsFunnel(), 10000000);
            }
            if(skuCount == null){ 
                skuCount = 0L;
            }
            if(!bloomFilter.mightContain(value)){ 
                bloomFilter.put(value);
                skuCount = skuCount + 1;
            }
            bloomState.update(bloomFilter);
            countState.update(skuCount);
            out.collect(countState.value());
        }
    }

我们使用 Guava 自带的 BloomFilter，每当来一条数据时，就检查 state 中的布隆过滤器中是否存在当前的 SKU，如果没有则初始化，如果有则数量加 1。

**基于 BitMap**

上面的 HyperLogLog 和 BloomFilter 虽然减少了存储但是丢失了精度， 这在某些业务场景下是无法被接受的。下面的这种方法不仅可以减少存储，而且还可以做到完全准确，那就是使用 BitMap。

> Bit-Map 的基本思想是用一个 bit 位来标记某个元素对应的 Value，而 Key 即是该元素。由于采用了 Bit 为单位来存储数据，因此可以大大节省存储空间。  
> 假设有这样一个需求：在 20 亿个随机整数中找出某个数 m 是否存在其中，并假设 32 位操作系统，4G 内存。在 Java 中，int 占 4 字节，1 字节 = 8 位（1 byte = 8 bit）  
> 如果每个数字用 int 存储，那就是 20 亿个 int，因而占用的空间约为 (2000000000\*4/1024/1024/1024)≈7.45G  
> 如果按位存储就不一样了，20 亿个数就是 20 亿位，占用空间约为 (2000000000/8/1024/1024/1024)≈0.233G

在使用 BitMap 算法前，如果你需要去重的对象不是数字，那么需要先转换成数字。例如，用户可以自己创造一个映射器，将需要去重的对象和数字进行映射，最简单的办法是，可以直接使用数据库维度表中自增 ID。

首先我们新增一个依赖：

<dependency>
       <groupId>org.roaringbitmap</groupId>
       <artifactId>RoaringBitmap</artifactId>
       <version>0.8.0</version>
    </dependency>

然后，我们还以商品的 SKU 的访问记录举例：

public class BitMapDistinct implements AggregateFunction<Long, Roaring64NavigableMap,Long> { 
    
        @Override
        public Roaring64NavigableMap createAccumulator() { 
            return new Roaring64NavigableMap();
        }
        @Override
        public Roaring64NavigableMap add(Long value, Roaring64NavigableMap accumulator) { 
            accumulator.add(value);
            return accumulator;
        }
        @Override
        public Long getResult(Roaring64NavigableMap accumulator) { 
            return accumulator.getLongCardinality();
        }
        @Override
        public Roaring64NavigableMap merge(Roaring64NavigableMap a, Roaring64NavigableMap b) { 
            return null;
        }
    }

在上述方法中，我们使用了 Roaring64NavigableMap，其是 BitMap 的一种实现，然后我们的数据是每次被访问的 SKU，把它直接添加到 Roaring64NavigableMap 中，最后通过 accumulator.getLongCardinality() 可以直接获取结果。

**基于外部数据库**

假如我们的业务场景非常复杂，并且数据量很大。为了防止无限制的状态膨胀，也不想维护庞大的 Flink 状态，我们可以采用外部存储的方式，比如可以选择使用 Redis 或者 HBase 存储数据，我们只需要设计好存储的 Key 即可。同时使用外部数据库进行存储，我们不需要关心 Flink 任务重启造成的状态丢失问题，但是有可能会出现因为重启恢复导致的数据多次发送，从而导致结果数据不准的问题。

# 第二十一讲 Flink 在实时计算平台和实时数据仓库中的作用 #

**基于 Flink 的实时计算平台**

大部分公司随着业务场景的不断丰富，同时在业界经过多年的实践检验，基于 Hadoop 的离线存储体系已经足够成熟。但是离线计算天然时效性不强，一般都是隔天级别的滞后，业务数据随着实践的推移，本身的价值就会逐渐减少。越来越多的场景需要使用实时计算，在这种背景下实时计算平台的需求应运而生。

**架构选型**

我们在第 03 课时“Flink 的编程模型与其他框架比较”中，提到过 Flink 自身独有的优势。

**首先在架构上**，Flink 采用了经典的主从模式，DataFlow Graph 与 Storm 形成的拓扑 Topology 结构类似，Flink 程序启动后，会根据用户的代码处理成 Stream Graph，然后优化成为 JobGraph，JobManager 会根据 JobGraph 生成 ExecutionGraph。ExecutionGraph 才是 Flink 真正能执行的数据结构，当很多个 ExecutionGraph 分布在集群中，就会形成一张网状的拓扑结构。

**其次在容错方面**，针对以前的 Spark Streaming 任务，我们可以配置对应的 checkpoint，也就是保存点（检查点）。当任务出现 failover 的时候，会从 checkpoint 重新加载，使得数据不丢失。但是这个过程会导致原来的数据重复处理，不能做到“只处理一次”的语义。Flink 基于两阶段提交实现了端到端的一次处理语义。

**在任务的反压上**，Flink 没有使用任何复杂的机制来解决反压问题，Flink 在数据传输过程中使用了分布式阻塞队列。我们知道在一个阻塞队列中，当队列满了以后发送者会被天然阻塞住，这种阻塞功能相当于给这个阻塞队列提供了反压的能力。

这些优势和特性，使得 Flink 在实时计算平台的搭建上占有一席之地。

**实时计算平台整体架构**

一般的实时计算平台的构成大都是以下几部分构成。

*  实时数据收集层

在实际业务中，大量的实时计算都是基于消息系统进行的数据收集和投递，这都离不开强大的消息中间件。目前业界使用最广的是 Kafka，另外一些重要的业务数据还会用到其他消息系统比如 RocketMQ 等。Kafka 因为高吞吐、低延迟的特性，特别适合大数量量、高 QPS 下的业务场景，而 RocketMQ 则在事务消息、一致性上有独特的优势。

*  实时计算层

Flink 在计算层同时支持流式及批量分析应用，这就是我们所说的批流一体。Flink 承担了数据的实时采集、实时计算和下游发送的角色。随着 Blink 的开源和一些其他实时产品的开源，支持可视化、SQL 化的开发模式已经越来越普及。

*  数据存储层

这里是我们的实时数据存储层，存储层除了传统 MySQL 等存储引擎以外，还会根据场景数据的不同存储在 Redis、HBase、OLAP 中。而这一层我个人认为最重要的技术选型则是 OLAP。OLAP 的技术选型直接制约着数据存储层和数据服务层的能力。关于 OLAP 的技术选型，可以参考[这里][Link 3]。

*  数据服务层

数据服务层会提供统一的对外查询、多维度的实时汇总，加上完善的租户和权限设计，能够支持多部门、多业务的数据需求。另外，基于数据服务层还会有数据的展示、大屏、指标可视化等。

**实时计算平台实际应用**

美团在公开发表的文章中提到，目前美团实时计算平台的架构组成：最底层是实时数据收集层，使用 Kafka 进行数据收集，支撑了大量的实时计算和离线数据拉取任务。

基于实时数据收集层之上，是基于 Flink 的实时计算层，美团选择了 Flink on Yarn 模式，并且选择了 Redis、HBase 和 ElasticSearch 作为数据存储。同时，美团还面向数据开发者进行作业托管、作业调优和诊断报警功能。

整体来看，美团的实时计算平台主要包含**作业管理**和**资源管理两个方面**的能力。基于作业管理上可以做到任务的发布、回滚、状态监控等，在资源管理上基于多租户的设计进行业务隔离，并且 Flink Job 采用 on Yarn 的模式，任务之间进行资源隔离。

根据公开的技术分享文档来看，目前美团点评的实时计算平台节点已经达到几千台，在资源的优化上需要做到自动扩容、缩容，另外实时计算任务和离线任务的混合部署也需要考虑到如何进行更细粒度资源的释放。

美团的实时计算平台图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 11]  
微博的实时计算平台是随着业务线的快速扩张，为了满足业务需求逐渐演变。在初期架构中，仅仅存在计算和存储两层，每次接入一个新的实时计算业务就要重新开发一遍，代码的利用率低下，接入成本很高。

基于上面的需求，微博开发了基于 Flink 的通用组件，用来快速支持实时数据的快速接入。从下到上共分为五层，分别是接入层、计算层、存储层、服务层和应用层。整体的架构图如下图所示：

微博的实时计算平台图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 12]  
基于微博的实时计算架构，数据从产生进入消息系统，通过 Flink 进行 ETL 进入存储层。根据业务不同的指标和数据需求，写入不同的存储中。统一查询服务则根据目前业务的需要，将查询服务微服务化，从数仓如 Redis、ElasticSearch、MySQL 等不同的存储中直接抽取。

微博的实时计算平台初期架构仅仅包含计算和存储两层，每个新的实时计算需求都要重新开发一遍，代码利用率低、重复开发。不同业务的不同需求对同一个指标的计算没有进行统一。随着数据量和业务的增长，前期架构的弊端开始显现，逐步发展成现在通用的计算架构。

在全新的计算架构下，微博基于 ClickHouse 进行多维度的计算来满足大数据量下的快速查询需求。数据分层上也借鉴了离线数仓的经验，构建了一套多层级的实时数仓服务，并且开发了多种形式的微服务对指标提取、数据聚合、数据质量、报警监控等进行支持。

**基于 Flink 的实时数据仓库**

我们在之前的课程中提过，Flink 的实际应用场景之一就是实时数据仓库。

**实时数仓背景**

传统的离线数据仓库将业务数据集中进行存储后，以固定的计算逻辑定时进行 ETL 和其他建模后产出报表等应用。离线数据仓库主要是构建 T+1 的离线数据，通过定时任务每天拉取增量数据，然后创建各个业务相关的主题维度数据，对外提供 T+1 的数据查询接口。

离线数据仓库 ETL 和实时数据仓库的差异图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 13]  
上图展示了离线数据仓库 ETL 和实时数据仓库的差异，可以看到离线数据仓库的计算和数据的实时性均较差。数据本身的价值随着时间的流逝会逐步减弱，因此数据发生后必须尽快地达到用户的手中，实时数仓的构建需求也应运而生。

实时数据仓库的建设是“数据智能 BI”必不可少的一环，也是大规模数据应用中必然面临的挑战。

**Flink 在实时数仓的优势**

Flink 在实时数仓和实时 ETL 中有天然的优势：

*  **状态管理**，实时数仓里面会进行很多的聚合计算，这些都需要对状态进行访问和管理，Flink 支持强大的状态管理；
 *  **丰富的 API**，Flink 提供极为丰富的多层次 API，包括 Stream API、Table API 及 Flink SQL；
 *  **生态完善**，实时数仓的用途广泛，Flink 支持多种存储（HDFS、ES 等）；
 *  **批流一体**，Flink 已经在将流计算和批计算的 API 进行统一。

**实时数仓的实际应用**

离线数据仓库的设计中，我们会把仓库结构分为不同的层次来存储不同的数据，大概可以分为：ODS 源数据、DWD 明细层、DWS 汇总层、ADM 应用层。在实时数据仓库的设计中也可以参考这个设计。

但是需要注意的是，在实时数据模型的处理方式上和离线有所区别。例如，明细层的汇总一般是基于 Flink 等接入 Kafka 消息进行关联的，维度表的数据一般会放在 HDFS、HBase 中作为明细层的补充。另外，在实时数据仓库中要选择不同的 OLAP 库来满足即席查询。

我们来看一下网易严选和美团的实时数据仓库的设计分别是怎样的。

**网易严选**

网易严选的实时数据仓库设计图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 14]  
如上图所示，网易严选的实时数据仓库 ODS 层主要是基于 Kafka 的事实数据，经过 Flink 处理后形成 DWD 明细层，在 DWD 层中会关联一些维度和历史数据，并且存入 Redis 中。在 DWS 层中会根据不同的业务场景有不同的存储，高并发查询和写入会基于 HBase 进行。如果你需要基于明细做不同维度的汇总那么就要在 GreenPulm 这个 OLAP 引擎中进行查询分析，另外一些维度较多的查询、排序等直接存储在 Redis 中供查询使用。

我们可以看出网易严选在建设实时数仓的主要考量是计算和存储。在计算上，网易严选选择了 Flink，主要是因为 Flink 的端到端的精确一次语义支持和容错特性。另外在存储方面，网易严选选择把 Flink 处理完的数据备份到 Kafka，并且根据业务场景和数据应用特点选择了不同的存储介质，例如一些高并发查询会基于 HBase 进行，常见的汇总指标会放入 MySQL 中直接使用。

在我们的实际业务场景下，明细和汇总数据会根据查询 QPS、维度的复杂程度等选择不同的介质，其中 HBase、OLAP、Redis 等都是常见存储介质，需要用户根据实际业务进行选择。

**美团**

下图是美团的实时数仓分层的架构模型图：

*  ODS 层，基于 MySQL Binlog 和 Kafka 的日志消息；
 *  明细层，基于事实数据关联成明细数据；
 *  汇总层，使用明细数据进行多维度的查询汇总；
 *  应用层，对外提供 HTTP、RPC 等查询服务。

美团的实时数仓分层架构模型图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 15]  
美团的 ODS 存放的主要是业务数据，其中大多是基于 MySQL 的 Binlog 和消息数据，这也是我们在实际业务中常用的方法。我们在建设实时数据仓库的过程中一个要求就是要和实际业务系统进行解耦，所以消息系统是我们的必然选择。

明细层是根据业务进行的划分，这一层的计算主要是基于 Flink 进行的，这一层承担着业务数据的解析、关联维表、明细存储的功能。

汇总层会基于明细数据进行再次关联和汇总，根据业务需要产出中间层和指标结果层。

最上层是同一对外服务的应用层，这层在我们的实际应用中非常重要。主要是根据外部系统的查询需要提供不同的查询服务，基于 HTTP、RPC 等的查询服务是常见选择，我们需要仔细评估访问的 QPS、查询负载等对接口进行限流、幂等、容错设计。并且需要进行严格的权限设计，防止数据泄露和非法访问。

# 第二十二讲 项目背景和整体架构设计 #

从这一课时开始我们进入实战课程的学习。本项目是一个模拟实时电商数据大屏，本课时先介绍该项目的背景、架构设计和技术选型。

**背景**

我们在第 01 课时“Flink 的应用场景和架构模型”中提到过，Flink 应用最广的一个场景便是实时计算大屏。每年的双十一、618 电商大促等，各大公司的实时数据战报和数据大屏是一道亮丽的风景线。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 16]  
实时大屏对数据有非常高的稳定性和精确性要求，特别是面向公众第三方的数据大屏，同时要求高吞吐、低延迟、极高的稳定性和绝对零误差。随时电商大促的成交记录一次次被刷新，背后是下单、支付、发货高达几万甚至十几万的峰值 QPS。

在面向实际运营的数据大屏中，需要提供高达几十种维度的数据，每秒的数据量高达千万甚至亿级别，这对于我们的实时计算架构提出了相当高的要求。那么我们的大屏背后的实时处理在这种数据量规模如何才能达到高吞吐、低延迟、极高的稳定性和绝对零误差的呢？

**技术选型和整体架构**

典型的实时计算大屏服务的背后技术架构图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 17]  
在上图的架构图中，涉及几个关键的技术选型，我们下面一一进行讲解。

**业务库 Binlog 同步利器——Canal**

我们的实时计算架构一般是基于业务数据进行的，但无论是实时计算大屏还是常规的数据分析报表，都不能影响业务的正常进行，所以这里需要引入消息中间件或增量同步框架 Canal。

我们生产环境中的业务数据绝大多数都是基于 MySQL 的，所以需要一个能够实时监控 MySQL 业务数据变化的工具。Canal 是阿里巴巴开源的数据库 Binlog 日志解析框架，主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 18]  
Canal 的原理也非常简单，它会伪装成一个数据库的从库，来读取 Binlog 并进行解析。关于 Canal 的更多资料，你可以参考[这里][Link 4]。

**解耦和海量数据支持——Kafka**

在实时大屏的技术架构下，我们的数据源绝大多数情况下都是消息。我们需要一个强大的消息中间件来支撑高达几十万 QPS，同时支持海量数据存储。

首先，我们为什么需要引入消息中间件？主要是下面三个目的：

*  同步变异步
 *  应用解耦
 *  流量削峰

在我们的架构中，为了和业务数据互相隔离，需要使用消息中间件进行解耦从而互不影响。另外在双十一等大促场景下，交易峰值通常出现在某一个时间段，这个时间段系统压力陡增，数据量暴涨，消息中间件还起到了削峰的作用。

为什么选择 Kafka？

Kafka 是最初由 Linkedin 公司开发，是一个分布式、高吞吐、多分区的消息中间件。Kafka 经过长时间的迭代和实践检验，因为其独特的优点已经成为目前主流的分布式消息引擎，经常被用作企业的消息总线、实时数据存储等。

Kafka 从众多的消息中间件中脱颖而出，主要是因为高吞吐、低延迟的特点；另外基于 Kafka 的生态越来越完善，各个实时处理框架包括 Flink 在消息处理上都会优先进行支持。在第 14 课时“Flink Exactly-once 实现原理解析”中提到了 Flink 和 Kafka 结合实现端到端精确一次语义的原理。

Kafka 作为大数据生态系统中已经必不可少的一员，主要的特性如下所示。

*  **高吞吐**：可以满足每秒百万级别消息的生产和消费，并且可以通过横向扩展，保证数据处理能力可以得到线性扩展。
 *  **低延迟**：以时间复杂度为 O(1) 的方式提供消息持久化能力，即使对 TB 级以上数据也能保证常数时间复杂度的访问性能。
 *  **高容错**：Kafka 允许集群的节点出现失败。
 *  **可靠性**：消息可以根据策略进行磁盘的持久化，并且读写效率都很高。
 *  **生态丰富**：Kafka 周边生态极其丰富，与各个实时处理框架结合紧密。

**实时计算服务——Flink**

Flink 在当前的架构中主要承担了消息消费、维表关联、消息发送等，我们在之前的课程中多次提到过 Flink 的优势，主要包括：

对于 Flink 的一些特点我们不做过多展开了。这里需要注意的是，Flink 在消费完成后一般会把计算结果数据发往三个方向：

*  **高度汇总**，高度汇总指标一般存储在 Redis、HBase 中供前端直接查询使用；
 *  **明细数据**，在一些场景下，我们的运营和业务人员需要查询明细数据，有一些明细数据极其重要，比如双十一派送的包裹中会有一些丢失和破损；
 *  **实时消息**，Flink 在计算完成后，有一个下游是发往消息系统，这里的作用主要是提供给其他业务复用；另外，在一些情况下，我们计算好明细数据也需要再次经过消息系统才能落库，将原来直接落库拆成两步，方便我们进行问题定位和排查。

**百花齐放——OLAP 数据库选择**

OLAP 的选择是当前实时架构中最有争议和最困难的。目前市面上主流的开源 OLAP 引擎包含但不限于：Hive、Hawq、Presto、Kylin、Impala、SparkSQL、Druid、Clickhouse、Greeplum 等，可以说目前没有一个引擎能在数据量，灵活程度和性能上做到完美，用户需要根据自己的需求进行选型。

我个人曾经在之前写过的[博客][Link 3]中用了两万字分析了目前市面上主流的 OLAP 数据库的选型问题，这里直接给出结论：

*  Hive、Hawq、Impala：基于 SQL on Hadoop
 *  Presto 和 Spark SQL 类似：基于内存解析 SQL 生成执行计划
 *  Kylin：用空间换时间、预计算
 *  Druid：数据实时摄入加实时计算
 *  ClickHouse：OLAP 领域的 HBase，单表查询性能优势巨大
 *  Greenpulm：OLAP 领域的 PostgreSQL

如果你的场景是基于 HDFS 的离线计算任务，那么 Hive、Hawq 和 Imapla 就是你的调研目标。

如果你的场景解决分布式查询问题，有一定的实时性要求，那么 Presto 和 SparkSQL 可能更符合你的期望。

如果你的汇总维度比较固定，实时性要求较高，可以通过用户配置的维度 + 指标进行预计算，那么不妨尝试 Kylin 和 Druid。

ClickHouse 则在单表查询性能上独领风骚，远超过其他的 OLAP 数据库。

Greenpulm 作为关系型数据库产品，性能可以随着集群的扩展线性增长，更加适合进行数据分析。

**实时大屏方案和指标**

在我们本次的案例中，将针对实时大屏中几个重要的指标进行计算，其中包括实时交易额、销售额排名 TOPN 等指标。

整个课程的设计包含以下几个部分：

*  实时数据的模拟
 *  Flink 消费 Kafka 数据开发
 *  Flink 中的业务逻辑开发
 *  Flink 计算结果写入 Redis 等
 *  其他

在这个案例中，我们还会从原理和源码层面讲解 Flink 消费 Kafka 的方式和注意事项等。

# 第二十三讲 Mock Kafka 消息并发送 #

本课时主要讲解 Kafka 的一些核心概念，以及模拟消息并发送。

**大数据消息中间件的王者——Kafka**

在上一课时中提过在实时计算的场景下，我们绝大多数的数据源都是消息系统。所以，一个强大的消息中间件来支撑高达几十万的 QPS，以及海量数据存储就显得极其重要。

Kafka 从众多的消息中间件中脱颖而出，主要是因为高吞吐、低延迟的特点；另外基于 Kafka 的生态越来越完善，各个实时处理框架包括 Flink 在消息处理上都会优先进行支持。在第 14 课时“Flink Exactly-once 实现原理解析”中提到 Flink 和 Kafka 结合实现端到端精确一次语义的原理。

Kafka 从众多的消息中间件中脱颖而出，已经成为大数据生态系统中必不可少的一员，主要的特性包括：

*  高吞吐
 *  低延迟
 *  高容错
 *  可靠性
 *  生态丰富

为了接下来更好地理解和使用 Kafka，我们首先来看一下 Kafka 中的核心概念和基本入门。

**Kafka 核心概念**

Kafka 是一个消息队列，生产者向消息队列中写入数据，消费者从队列中获取数据并进行消费。作为一个企业级的消息中间件，Kafka 会支持庞大的业务，不同的业务会有多个队列，我们用 Topic 来给队列命名，在使用 Kafka 时必须指定 Topic。

我们可以认为一个 Topic 就是一个队列，每个 Topic 又会被分成多个 Partition，这样做是为了横向扩展，提高吞吐量。

Kafka 中每个 Partition 都对应一个 Broker，一个 Broker 可以管理多个 Partition。举个例子，假如 Kafka 的某个 Topic 有 10 个 Partition、2 个 Broker，那么每个 Broker 就会管理 5 个 Partition。我们可以把 Partition 简单理解为一个文件，在接收生产者的数据时，需要将数据动态追加到 Partition 上。

生产者会决定将数据写入哪个 Partition，消费者自己维护消费数据的位置，我们称为 Offset。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 19]  
同时，Kafka 提供了时间策略对过期的消息进行处理。

Kafka 的每个消费者都有一个消费组来进行标识，同一个消费组的不同实例分布在多个进程或者多个机器上。

Kafka 的源数据存储在 ZooKeeper 中，其中包含 Broker、Topic、Partition 等信息。在 0.8 版本之前，Kafka 还会将消费的 Offset 存储在 ZooKeeper 中。

此外，ZooKeeper 还负责集群的 Broker 选举，以及所有 Topic 的 Partition 副本信息等。

**Kafka 连接 Flink**

我们在第 12 课时“Flink 常用的 Source 和 Connector”中提过，Flink 中支持了比较丰富的用来连接第三方的连接器，Kafka Connector 是 Flink 支持的各种各样的连接器中比较完善的之一。

> Flink 提供了专门的 Kafka 连接器，向 Kafka Topic 中读取或者写入数据。Flink Kafka Consumer 集成了 Flink 的 Checkpoint 机制，可提供 exactly-once 的处理语义。为此，Flink 并不完全依赖于跟踪 Kafka 消费组的偏移量，而是在内部跟踪和检查偏移量。

同时也提过，我们在使用 Kafka 连接器时需要引用相对应的 Jar 包依赖。对于某些连接器比如 Kafka 是有版本要求的，一定要去[官方网站][Link 5]找到对应的依赖版本。

我在下表中给出了不同版本的 Kafka，以及对应的 Connector 关系：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 20]  
**Kafka 本地环境搭建**

我们在本地环境搭建一个 Kafka\_2.11-2.1.0 版本的 Kafka 单机环境，然后模拟一些数据写入到队列中。

我们可以在[这里][Link 6]下载对应版本的 Kafka，把压缩包进行解压，然后使用下面的命令启动单机版本的 Kafka。

解压：

> tar -xzf kafka_2.11-2.1.0.tgz
    > cd kafka_2.11-2.1.0

启动 ZooKeeper 和 Kafka Server：

启动ZK：nohup bin/zookeeper-server-start.sh config/zookeeper.properties  &
    启动Server: 
    nohup bin/kafka-server-start.sh config/server.properties &

创建一个名为 test 的 Topic：

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

**Kafka Producer**

首先我们需要新增一个依赖，然后向名为 test 的 Topic 中写入数据。

新增 Maven 依赖：

<dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-connector-kafka_2.11</artifactId>
      <version>1.10.0</version>
    </dependency>

向 test 这个 Topic 中写入数据：

public class KafkaProducer { 
        public static void main(String[] args) throws Exception{ 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            env.enableCheckpointing(5000);
            DataStreamSource<String> text = env.addSource(new MyNoParalleSource()).setParallelism(1);
            Properties properties = new Properties();
            properties.setProperty("bootstrap.servers", "127.0.0.1:9092");
            // 2.0 配置 KafkaProducer
            FlinkKafkaProducer<String> producer = new FlinkKafkaProducer<String>(
                    "127.0.0.1:9092", //broker 列表
                    "test",           //topic
                    new SimpleStringSchema()); // 消息序列化
            
            //写入 Kafka 时附加记录的事件时间戳
            producer.setWriteTimestampToKafka(true);
            text.addSink(producer);
            env.execute();
        }
    }

需要注意的是，我们这里使用了一个自定义的 MyNoParalleSource 类，该类使用了 Flink 提供的自定义 Source 方法，该方法会源源不断地产生一些测试数据，代码如下：

public class MyNoParalleSource implements SourceFunction<String> { 
        //private long count = 1L;
        private boolean isRunning = true;
        /** * 主要的方法 * 启动一个source * 大部分情况下，都需要在这个run方法中实现一个循环，这样就可以循环产生数据了 * * @param ctx * @throws Exception */
        @Override
        public void run(SourceContext<String> ctx) throws Exception { 
            while(isRunning){ 
                //图书的排行榜
                List<String> books = new ArrayList<>();
                books.add("Pyhton从入门到放弃");//10
                books.add("Java从入门到放弃");//8
                books.add("Php从入门到放弃");//5
                books.add("C++从入门到放弃");//3
                books.add("Scala从入门到放弃");
                int i = new Random().nextInt(5);
                ctx.collect(books.get(i));
                //每2秒产生一条数据
                Thread.sleep(2000);
            }
        }
        //取消一个cancel的时候会调用的方法
        @Override
        public void cancel() { 
            isRunning = false;
        }
    }

我们在后面会使用这个方法来模拟生产中的订单数据，并进行接续处理；然后通过下面的命令就可以查看本地 Kafka 的 test 这个 Topic 中的数据：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 21]  
至此，我们就成功地向 Kafka 中写入数据了。

**源码解析**

FlinkKafkaProducer 的代码十分简洁，首先继承了 TwoPhaseCommitSinkFunction，在第 14 课时“Flink Exactly-once 实现原理解析”中详细讲解过，这个类是 Flink 和 Kafka 结合实现精确一次处理语义的关键。

FlinkProducer 提供了 6 种构造方法，我们可以根据需要选择不同的构造函数：

public FlinkKafkaProducer011(
                String brokerList, 
                String topicId, 
                SerializationSchema<IN> serializationSchema);
     
    public FlinkKafkaProducer011(
                String topicId, 
                SerializationSchema<IN> serializationSchema, 
                Properties producerConfig);
     
    public FlinkKafkaProducer011(
                String topicId, 
                SerializationSchema<IN> serializationSchema, 
                Properties producerConfig, 
                Optional<FlinkKafkaPartitioner<IN>> customPartitioner);
     
    public FlinkKafkaProducer011(
                String brokerList, 
                String topicId, 
                KeyedSerializationSchema<IN> serializationSchema);
     
    public FlinkKafkaProducer011(
                String topicId, 
                KeyedSerializationSchema<IN> serializationSchema, 
                Properties producerConfig);
     
    public FlinkKafkaProducer011(
                String topicId, 
                KeyedSerializationSchema<IN> serializationSchema, 
                Properties producerConfig, 
                Semantic semantic);
     
    public FlinkKafkaProducer011(
                String defaultTopicId, 
                KeyedSerializationSchema<IN> serializationSchema, 
                Properties producerConfig, 
                Optional<FlinkKafkaPartitioner<IN>> customPartitioner);
     
    public FlinkKafkaProducer011(
    			String defaultTopicId,
    			KeyedSerializationSchema<IN> serializationSchema,
    			Properties producerConfig,
    			Optional<FlinkKafkaPartitioner<IN>> customPartitioner,
    			Semantic semantic,
    			int kafkaProducersPoolSize);

这里有个特别需要注意的属性：FlinkKafkaPartitioner，这个类定义了数据写入 Kafka 的规则，如果用户没有指定，则会默认 FlinkFixedPartitioner，核心处理逻辑如下：

public class FlinkFixedPartitioner<T> extends FlinkKafkaPartitioner<T> { 
    ...
       @Override
       public int partition(T record, byte[] key, byte[] value, String targetTopic, int[] partitions) { 
          Preconditions.checkArgument(
             partitions != null && partitions.length > 0,
             "Partitions of the target topic is empty.");
          return partitions[parallelInstanceId % partitions.length];
       }
    ...
    }

此外，FlinkProducer 还封装了 beginTransaction、preCommit、commit、abort 等方法，这几个方法便是实现精确一次处理语义的关键。

# 第二十四讲 Flink 消费 Kafka 数据业务开发 #

在上一课时中我们提过在实时计算的场景下，绝大多数的数据源都是消息系统，而 Kafka 从众多的消息中间件中脱颖而出，主要是因为高吞吐、低延迟的特点；同时也讲了 Flink 作为生产者像 Kafka 写入数据的方式和代码实现。这一课时我们将从以下几个方面介绍 Flink 消费 Kafka 中的数据方式和源码实现。

**Flink 如何消费 Kafka**

Flink 在和 Kafka 对接的过程中，跟 Kafka 的版本是强相关的。上一课时也提到了，我们在使用 Kafka 连接器时需要引用相对应的 Jar 包依赖，对于某些连接器比如 Kafka 是有版本要求的，一定要去[官方网站][Link 5]找到对应的依赖版本。

我们本地的 Kafka 版本是 2.1.0，所以需要对应的类是 FlinkKafkaConsumer。首先需要在 pom.xml 中引入 jar 包依赖：

<dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-connector-kafka_2.11</artifactId>
      <version>1.10.0</version>
    </dependency>

下面将对 Flink 消费 Kafka 数据的方式进行分类讲解。

**消费单个 Topic**

上一课时我们在本地搭建了 Kafka 环境，并且手动创建了名为 test 的 Topic，然后向名为 test 的 Topic 中写入了数据。

那么现在我们要消费这个 Topic 中的数据，该怎么做呢？

public static void main(String[] args) throws Exception { 
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.enableCheckpointing(5000);
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "127.0.0.1:9092");
        // 如果你是0.8版本的Kafka，需要配置
        //properties.setProperty("zookeeper.connect", "localhost:2181");
        //设置消费组
        properties.setProperty("group.id", "group_test");
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("test", new SimpleStringSchema(), properties);
        //设置从最早的ffset消费
        consumer.setStartFromEarliest();
        //还可以手动指定相应的 topic, partition，offset,然后从指定好的位置开始消费
        //HashMap<KafkaTopicPartition, Long> map = new HashMap<>();
        //map.put(new KafkaTopicPartition("test", 1), 10240L);
        //假如partition有多个，可以指定每个partition的消费位置
        //map.put(new KafkaTopicPartition("test", 2), 10560L);
        //然后各个partition从指定位置消费
        //consumer.setStartFromSpecificOffsets(map);
        env.addSource(consumer).flatMap(new FlatMapFunction<String, String>() { 
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception { 
                System.out.println(value);
            }
        });
        env.execute("start consumer...");
    }

在设置消费 Kafka 中的数据时，可以显示地指定从某个 Topic 的每一个 Partition 中进行消费。

**消费多个 Topic**

我们的业务中会有这样的情况，同样的数据根据类型不同发送到了不同的 Topic 中，比如线上的订单数据根据来源不同分别发往移动端和 PC 端两个 Topic 中。但是我们不想把同样的代码复制一份，需重新指定一个 Topic 进行消费，这时候应该怎么办呢？

Properties properties = new Properties();
    properties.setProperty("bootstrap.servers", "127.0.0.1:9092");
    // 如果你是0.8版本的Kafka，需要配置
    //properties.setProperty("zookeeper.connect", "localhost:2181");
    //设置消费组
    properties.setProperty("group.id", "group_test");
    FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("test", new SimpleStringSchema(), properties);
    ArrayList<String> topics = new ArrayList<>();
    topics.add("test_A");
    topics.add("test_B");
    // 传入一个 list，完美解决了这个问题
    FlinkKafkaConsumer<Tuple2<String, String>> consumer = new FlinkKafkaConsumer<>(topics, new SimpleStringSchema(), properties);
    ...

我们可以传入一个 list 来解决消费多个 Topic 的问题，如果用户需要区分两个 Topic 中的数据，那么需要在发往 Kafka 中数据新增一个字段，用来区分来源。

**消息序列化**

我们在上述消费 Kafka 消息时，都默认指定了消息的序列化方式，即 SimpleStringSchema。这里需要注意的是，在我们使用 SimpleStringSchema 的时候，返回的结果中只有原数据，没有 topic、parition 等信息，这时候可以自定义序列化的方式来实现自定义返回数据的结构。

public class CustomDeSerializationSchema implements KafkaDeserializationSchema<ConsumerRecord<String, String>> { 
        //是否表示流的最后一条元素,设置为false，表示数据会源源不断地到来
        @Override
        public boolean isEndOfStream(ConsumerRecord<String, String> nextElement) { 
            return false;
        }
        //这里返回一个ConsumerRecord<String,String>类型的数据，除了原数据还包括topic，offset，partition等信息
        @Override
        public ConsumerRecord<String, String> deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception { 
            return new ConsumerRecord<String, String>(
                    record.topic(),
                    record.partition(),
                    record.offset(),
                    new String(record.key()),
                    new String(record.value())
            );
        }
        //指定数据的输入类型
        @Override
        public TypeInformation<ConsumerRecord<String, String>> getProducedType() { 
            return TypeInformation.of(new TypeHint<ConsumerRecord<String, String>>(){ });
        }
    }

这里自定义了 CustomDeSerializationSchema 信息，就可以直接使用了。

**Parition 和 Topic 动态发现**

在很多场景下，随着业务的扩展，我们需要对 Kafka 的分区进行扩展，为了防止新增的分区没有被及时发现导致数据丢失，消费者必须要感知 Partition 的动态变化，可以使用 FlinkKafkaConsumer 的动态分区发现实现。

我们只需要指定下面的配置，即可打开动态分区发现功能：每隔 10ms 会动态获取 Topic 的元数据，对于新增的 Partition 会自动从最早的位点开始消费数据。

properties.setProperty(FlinkKafkaConsumerBase.KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS, "10");

如果业务场景需要我们动态地发现 Topic，可以指定 Topic 的正则表达式：

FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(Pattern.compile("^test_([A-Za-z0-9]*)$"), new SimpleStringSchema(), properties);

**Flink 消费 Kafka 设置 offset 的方法**

Flink 消费 Kafka 需要指定消费的 offset，也就是偏移量。Flink 读取 Kafka 的消息有五种消费方式：

*  指定 Topic 和 Partition
 *  从最早位点开始消费
 *  从指定时间点开始消费
 *  从最新的数据开始消费
 *  从上次消费位点开始消费

/** * Flink从指定的topic和parition中指定的offset开始 */
    Map<KafkaTopicPartition, Long> offsets = new HashedMap();
    offsets.put(new KafkaTopicPartition("test", 0), 10000L);
    offsets.put(new KafkaTopicPartition("test", 1), 20000L);
    offsets.put(new KafkaTopicPartition("test", 2), 30000L);
    consumer.setStartFromSpecificOffsets(offsets);
    /** * Flink从topic中最早的offset消费 */
    consumer.setStartFromEarliest();
    /** * Flink从topic中指定的时间点开始消费 */
    consumer.setStartFromTimestamp(1559801580000l);
    /** * Flink从topic中最新的数据开始消费 */
    consumer.setStartFromLatest();
    /** * Flink从topic中指定的group上次消费的位置开始消费，所以必须配置group.id参数 */
    consumer.setStartFromGroupOffsets();

**源码解析**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 22]  
从上面的类图可以看出，FlinkKafkaConsumer 继承了 FlinkKafkaConsumerBase，而 FlinkKafkaConsumerBase 最终是对 SourceFunction 进行了实现。

整体的流程：FlinkKafkaConsumer 首先创建了 KafkaFetcher 对象，然后 KafkaFetcher 创建了 KafkaConsumerThread 和 Handover，KafkaConsumerThread 负责直接从 Kafka 中读取 msg，并交给 Handover，然后 Handover 将 msg 传递给 KafkaFetcher.emitRecord 将消息发出。

因为 FlinkKafkaConsumerBase 实现了 RichFunction 接口，所以当程序启动的时候，会首先调用 FlinkKafkaConsumerBase.open 方法：

public void open(Configuration configuration) throws Exception { 
       // 指定offset的提交方式
       this.offsetCommitMode = OffsetCommitModes.fromConfiguration(
             getIsAutoCommitEnabled(),
             enableCommitOnCheckpoints,
             ((StreamingRuntimeContext) getRuntimeContext()).isCheckpointingEnabled());
       // 创建分区发现器
       this.partitionDiscoverer = createPartitionDiscoverer(
             topicsDescriptor,
             getRuntimeContext().getIndexOfThisSubtask(),
             getRuntimeContext().getNumberOfParallelSubtasks());
       this.partitionDiscoverer.open();
       subscribedPartitionsToStartOffsets = new HashMap<>();
       final List<KafkaTopicPartition> allPartitions = partitionDiscoverer.discoverPartitions();
       if (restoredState != null) { 
          for (KafkaTopicPartition partition : allPartitions) { 
             if (!restoredState.containsKey(partition)) { 
                restoredState.put(partition, KafkaTopicPartitionStateSentinel.EARLIEST_OFFSET);
             }
          }
          for (Map.Entry<KafkaTopicPartition, Long> restoredStateEntry : restoredState.entrySet()) { 
             if (!restoredFromOldState) { 
               
                if (KafkaTopicPartitionAssigner.assign(
                   restoredStateEntry.getKey(), getRuntimeContext().getNumberOfParallelSubtasks())
                      == getRuntimeContext().getIndexOfThisSubtask()){ 
                   subscribedPartitionsToStartOffsets.put(restoredStateEntry.getKey(), restoredStateEntry.getValue());
                }
             } else { 
               subscribedPartitionsToStartOffsets.put(restoredStateEntry.getKey(), restoredStateEntry.getValue());
             }
          }
          if (filterRestoredPartitionsWithCurrentTopicsDescriptor) { 
             subscribedPartitionsToStartOffsets.entrySet().removeIf(entry -> { 
                if (!topicsDescriptor.isMatchingTopic(entry.getKey().getTopic())) { 
                   LOG.warn(
                      "{} is removed from subscribed partitions since it is no longer associated with topics descriptor of current execution.",
                      entry.getKey());
                   return true;
                }
                return false;
             });
          }
          LOG.info("Consumer subtask {} will start reading {} partitions with offsets in restored state: {}",
             getRuntimeContext().getIndexOfThisSubtask(), subscribedPartitionsToStartOffsets.size(), subscribedPartitionsToStartOffsets);
       } else { 
        
          switch (startupMode) { 
             case SPECIFIC_OFFSETS:
                if (specificStartupOffsets == null) { 
                   throw new IllegalStateException(
                      "Startup mode for the consumer set to " + StartupMode.SPECIFIC_OFFSETS +
                         ", but no specific offsets were specified.");
                }
                for (KafkaTopicPartition seedPartition : allPartitions) { 
                   Long specificOffset = specificStartupOffsets.get(seedPartition);
                   if (specificOffset != null) { 
                                     subscribedPartitionsToStartOffsets.put(seedPartition, specificOffset - 1);
                   } else { 
                   subscribedPartitionsToStartOffsets.put(seedPartition, KafkaTopicPartitionStateSentinel.GROUP_OFFSET);
                   }
                }
                break;
             case TIMESTAMP:
                if (startupOffsetsTimestamp == null) { 
                   throw new IllegalStateException(
                      "Startup mode for the consumer set to " + StartupMode.TIMESTAMP +
                         ", but no startup timestamp was specified.");
                }
                for (Map.Entry<KafkaTopicPartition, Long> partitionToOffset
                      : fetchOffsetsWithTimestamp(allPartitions, startupOffsetsTimestamp).entrySet()) { 
                   subscribedPartitionsToStartOffsets.put(
                      partitionToOffset.getKey(),
                      (partitionToOffset.getValue() == null)
                          KafkaTopicPartitionStateSentinel.LATEST_OFFSET
                            : partitionToOffset.getValue() - 1);
                }
                break;
             default:
                for (KafkaTopicPartition seedPartition : allPartitions) { 
                   subscribedPartitionsToStartOffsets.put(seedPartition, startupMode.getStateSentinel());
                }
          }
          if (!subscribedPartitionsToStartOffsets.isEmpty()) { 
             switch (startupMode) { 
                case EARLIEST:
                   LOG.info("Consumer subtask {} will start reading the following {} partitions from the earliest offsets: {}",
                      getRuntimeContext().getIndexOfThisSubtask(),
                      subscribedPartitionsToStartOffsets.size(),
                      subscribedPartitionsToStartOffsets.keySet());
                   break;
                case LATEST:
                   LOG.info("Consumer subtask {} will start reading the following {} partitions from the latest offsets: {}",
                      getRuntimeContext().getIndexOfThisSubtask(),
                      subscribedPartitionsToStartOffsets.size(),
                      subscribedPartitionsToStartOffsets.keySet());
                   break;
                case TIMESTAMP:
                   LOG.info("Consumer subtask {} will start reading the following {} partitions from timestamp {}: {}",
                      getRuntimeContext().getIndexOfThisSubtask(),
                      subscribedPartitionsToStartOffsets.size(),
                      startupOffsetsTimestamp,
                      subscribedPartitionsToStartOffsets.keySet());
                   break;
                case SPECIFIC_OFFSETS:
                   LOG.info("Consumer subtask {} will start reading the following {} partitions from the specified startup offsets {}: {}",
                      getRuntimeContext().getIndexOfThisSubtask(),
                      subscribedPartitionsToStartOffsets.size(),
                      specificStartupOffsets,
                      subscribedPartitionsToStartOffsets.keySet());
                   List<KafkaTopicPartition> partitionsDefaultedToGroupOffsets = new ArrayList<>(subscribedPartitionsToStartOffsets.size());
                   for (Map.Entry<KafkaTopicPartition, Long> subscribedPartition : subscribedPartitionsToStartOffsets.entrySet()) { 
                      if (subscribedPartition.getValue() == KafkaTopicPartitionStateSentinel.GROUP_OFFSET) { 
                         partitionsDefaultedToGroupOffsets.add(subscribedPartition.getKey());
                      }
                   }
                   if (partitionsDefaultedToGroupOffsets.size() > 0) { 
                      LOG.warn("Consumer subtask {} cannot find offsets for the following {} partitions in the specified startup offsets: {}" +
                            "; their startup offsets will be defaulted to their committed group offsets in Kafka.",
                         getRuntimeContext().getIndexOfThisSubtask(),
                         partitionsDefaultedToGroupOffsets.size(),
                         partitionsDefaultedToGroupOffsets);
                   }
                   break;
                case GROUP_OFFSETS:
                   LOG.info("Consumer subtask {} will start reading the following {} partitions from the committed group offsets in Kafka: {}",
                      getRuntimeContext().getIndexOfThisSubtask(),
                      subscribedPartitionsToStartOffsets.size(),
                      subscribedPartitionsToStartOffsets.keySet());
             }
          } else { 
             LOG.info("Consumer subtask {} initially has no partitions to read from.",
                getRuntimeContext().getIndexOfThisSubtask());
          }
       }
    }

对 Kafka 中的 Topic 和 Partition 的数据进行读取的核心逻辑都在 run 方法中：

public void run(SourceContext<T> sourceContext) throws Exception { 
       if (subscribedPartitionsToStartOffsets == null) { 
          throw new Exception("The partitions were not set for the consumer");
       }
       this.successfulCommits = this.getRuntimeContext().getMetricGroup().counter(COMMITS_SUCCEEDED_METRICS_COUNTER);
       this.failedCommits =  this.getRuntimeContext().getMetricGroup().counter(COMMITS_FAILED_METRICS_COUNTER);
       final int subtaskIndex = this.getRuntimeContext().getIndexOfThisSubtask();
       this.offsetCommitCallback = new KafkaCommitCallback() { 
          @Override
          public void onSuccess() { 
             successfulCommits.inc();
          }
          @Override
          public void onException(Throwable cause) { 
             LOG.warn(String.format("Consumer subtask %d failed async Kafka commit.", subtaskIndex), cause);
             failedCommits.inc();
          }
       };
    
       if (subscribedPartitionsToStartOffsets.isEmpty()) { 
          sourceContext.markAsTemporarilyIdle();
       }
       LOG.info("Consumer subtask {} creating fetcher with offsets {}.",
          getRuntimeContext().getIndexOfThisSubtask(), subscribedPartitionsToStartOffsets);
      
       this.kafkaFetcher = createFetcher(
             sourceContext,
             subscribedPartitionsToStartOffsets,
             periodicWatermarkAssigner,
             punctuatedWatermarkAssigner,
             (StreamingRuntimeContext) getRuntimeContext(),
             offsetCommitMode,
             getRuntimeContext().getMetricGroup().addGroup(KAFKA_CONSUMER_METRICS_GROUP),
             useMetrics);
       if (!running) { 
          return;
       }
       if (discoveryIntervalMillis == PARTITION_DISCOVERY_DISABLED) { 
          kafkaFetcher.runFetchLoop();
       } else { 
          runWithPartitionDiscovery();
       }
    }

**Flink 消费 Kafka 数据代码**

上面介绍了 Flink 消费 Kafka 的方式，以及消息序列化的方式，同时介绍了分区和 Topic 的动态发现方法，那么回到我们的项目中来，消费 Kafka 数据的完整代码如下：

public class KafkaConsumer { 
        public static void main(String[] args) throws Exception { 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            env.enableCheckpointing(5000);
            Properties properties = new Properties();
            properties.setProperty("bootstrap.servers", "127.0.0.1:9092");
            //设置消费组
            properties.setProperty("group.id", "group_test");
            properties.setProperty(FlinkKafkaConsumerBase.KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS, "10");
            FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("test", new SimpleStringSchema(), properties);
            //设置从最早的ffset消费
            consumer.setStartFromEarliest();
            env.addSource(consumer).flatMap(new FlatMapFunction<String, String>() { 
                @Override
                public void flatMap(String value, Collector<String> out) throws Exception { 
                    System.out.println(value);
                }
            });
            env.execute("start consumer...");
        }
    }

我们可以直接右键运行代码，在控制台中可以看到数据的正常打印，如下图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 23]  
通过代码可知，我们之前发往 Kafka 的消息被完整地打印出来了。

# 第二十五讲 Flink 中 watermark 的定义和使用 #

第 08 课时我们提过窗口和时间的概念，Flink 框架支持事件时间、摄入时间和处理时间三种。Watermark（水印）的出现是用于处理数据从 Source 产生，再到转换和输出，在这个过程中由于网络和反压的原因导致了消息乱序问题。

那么在实际的开发过程中，如何正确地使用 Watermark 呢？

**使用 Watermark 必知必会**

**Watermark 和事件时间**

事件时间（Event Time）是数据产生的时间，这个时间一般在数据中自带，由消息的生产者生成。例如，我们的上游是 Kafka 消息，那么每个生成的消息中自带一个时间戳代表该条数据的产生时间，这个时间是固定的，从数据的诞生开始就一直携带。所以，我们在处理消息乱序的情况时，会用 EventTime 和 Watermark 进行配合使用。

我们只需要一行代码，就可以在代码中指定 Flink 系统使用的时间类型为 EventTime：

env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

那么为什么不用处理时间（Processing Time）和摄入时间（Ingestion Time）呢？

处理时间（Processing Time）指的是数据被 Flink 框架处理时机器的系统时间，这个时间本身存在不确定性，比如因为网络延迟等原因。

摄入时间（Ingestion Time）理论上处于事件时间（Event Time）和处理时间（Processing Time）之间，可以用来防止 Flink 内部处理数据发生乱序的情况，但是无法解决数据进入 Flink 之前的乱序行为。

所以我们一般都会用 EventTime、WaterMark 和窗口配合使用来解决消息的乱序和延迟问题。

**水印的本质是时间戳**

水印的本质是一个一个的时间戳，这个时间戳存在 DataStream 的数据流中，Watermark 的生成方式有两种：

*  AssignerWithPeriodicWatermarks 生成周期水印，周期默认的时间是 200ms；
 *  AssignerWithPunctuatedWatermarks 按需生成水印。

当 Flink 系统中出现了一个 Watermark T，那么就意味着 EventTime <= T 的数据都已经到达。当 Wartermark T 通过窗口后，后续到来的迟到数据就会被丢弃。

**窗口触发和乱序时间**

Flink 在用时间 + 窗口 + 水印来解决实际生产中的数据乱序问题，有如下的触发条件：

*  watermark 时间 >= window\_end\_time；
 *  在 \[window\_start\_time,window\_end\_time) 中有数据存在，这个窗口是左闭右开的。

但是有些业务场景需要我们等待一段时间，也就是接受一定范围的迟到数据，此时 allowedLateness 的设置就显得尤为重要。简单地说，allowedLateness 的设置就是对于那些水印通过窗口的结束时间后，还允许等待一段时间。

如果业务中的实际数据因为网络原因，乱序现象非常严重，allowedLateness 允许迟到的时间如果设置太小，则会导致很多次极少量数据触发窗口计算，严重影响数据的正确性。

**Flink 消费 Kafka 保证消息有序**

我们在第 23 课时“Mock Kafka 消息并发送”中提过，可以认为 Kafka 中的一个 Topic 就是一个队列，每个 Topic 又会被分成多个 Partition，每个 Partition 中的消息是有序的。但是有的业务场景需要我们保障所有 Partition 中的消息有序，一般情况下需要把 Partition 的个数设置为一个，但这种情况是不能接受的，会严重影响数据的吞吐量。

但是，Flink 消费 Kafka 时可以做到数据的全局有序，也可以多个 Partition 并发消费，这就是 Flink 中的 Kafka-partition-aware 特性。

我们在使用这种特性生成水印时，水印会在 Flink 消费 Kafka 的消费端生成，并且每个分区的时间戳严格升序。当数据进行 Shuffle 时，水印的合并机制会产生全局有序的水印。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 24]  
我们从上图中可以看出，每个生成的水印是如何在多个分区的数据中进行传递的。

代码实现如下：

FlinkKafkaConsumer09<MyType> kafkaSource = new FlinkKafkaConsumer09<>("topic", schema, props);
    kafkaSource.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<MyType>() { 
        @Override
        public long extractAscendingTimestamp(MyType element) { 
            return element.eventTimestamp();
        }
    });
    DataStream<MyType> stream = env.addSource(kafkaSource);

**Flink 预定义的时间戳提取器和水印发射器**

Flink 本身提供了两个预定义实现类去生成水印：

*  AscendingTimestampExtractor 时间戳递增
 *  BoundedOutOfOrdernessTimestampExtractor 处理乱序消息和延迟时间

**AscendingTimestampExtractor 递增时间戳提取器**

AscendingTimestampExtractor 是周期性生成水印的一个简单实现，这种方式会产生严格递增的水印。它的实现如下：

public abstract class AscendingTimestampExtractor<T> implements AssignerWithPeriodicWatermarks<T> { 
    ...
       public AscendingTimestampExtractor<T> withViolationHandler(MonotonyViolationHandler handler) { 
          this.violationHandler = requireNonNull(handler);
          return this;
       }
       @Override
       public final long extractTimestamp(T element, long elementPrevTimestamp) { 
          final long newTimestamp = extractAscendingTimestamp(element);
          if (newTimestamp >= this.currentTimestamp) { 
             this.currentTimestamp = newTimestamp;
             return newTimestamp;
          } else { 
             violationHandler.handleViolation(newTimestamp, this.currentTimestamp);
             return newTimestamp;
          }
       }
       @Override
       public final Watermark getCurrentWatermark() { 
          return new Watermark(currentTimestamp == Long.MIN_VALUE ? Long.MIN_VALUE : currentTimestamp - 1);
       }
    ...
    }

该种水印的生成方式适用于那些数据本身的时间戳在每个并行的任务中是单调递增的，例如，我们上面使用 AscendingTimestampExtractor 处理 Kafka 多个 Partition 的情况。

一个简单的案例如下所示：

DataStream<MyEvent> stream = ...
    DataStream<MyEvent> withTimestampsAndWatermarks =
        stream.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<MyEvent>() { 
            @Override
            public long extractAscendingTimestamp(MyEvent element) { 
                return element.getCreationTime();
            }
    });

**BoundedOutOfOrdernessTimestampExtractor 允许特定数量延迟的提取器**

我们在上面提过有些业务场景需要等待一段时间，也就是接受一定范围的迟到数据，此时 allowedLateness 的设置就显得尤为重要。这种提取器也是周期性生成水印的实现，接受 allowedLateness 作为参数。

它的实现如下：

public abstract class BoundedOutOfOrdernessTimestampExtractor<T> implements AssignerWithPeriodicWatermarks<T> { 
    ...
       private final long maxOutOfOrderness;
       public BoundedOutOfOrdernessTimestampExtractor(Time maxOutOfOrderness) { 
          if (maxOutOfOrderness.toMilliseconds() < 0) { 
             throw new RuntimeException("Tried to set the maximum allowed " +
                "lateness to " + maxOutOfOrderness + ". This parameter cannot be negative.");
          }
          this.maxOutOfOrderness = maxOutOfOrderness.toMilliseconds();
          this.currentMaxTimestamp = Long.MIN_VALUE + this.maxOutOfOrderness;
       }
       public long getMaxOutOfOrdernessInMillis() { 
          return maxOutOfOrderness;
       }
    ...
       @Override
       public final Watermark getCurrentWatermark() { 
          // this guarantees that the watermark never goes backwards.
          long potentialWM = currentMaxTimestamp - maxOutOfOrderness;
          if (potentialWM >= lastEmittedWatermark) { 
             lastEmittedWatermark = potentialWM;
          }
          return new Watermark(lastEmittedWatermark);
       }
       @Override
       public final long extractTimestamp(T element, long previousElementTimestamp) { 
          long timestamp = extractTimestamp(element);
          if (timestamp > currentMaxTimestamp) { 
             currentMaxTimestamp = timestamp;
          }
          return timestamp;
       }
    }

BoundedOutOfOrdernessTimestampExtractor 的构造器接收 maxOutOfOrderness 这个参数，该参数是指定我们接收的消息允许滞后的最大时间。

**案例**

下面是一个接收 Kafka 消息进行处理，自定义窗口和水印的案例：

public class WindowWaterMark { 
    
        public static void main(String[] args) throws Exception { 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
            //设置为eventtime事件类型
            env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
            //设置水印生成时间间隔100ms
            env.getConfig().setAutoWatermarkInterval(100);
            DataStream<String> dataStream = env
                    .socketTextStream("127.0.0.1", 9000)
                    .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<String>() { 
                        private Long currentTimeStamp = 0L;
                        //设置允许乱序时间
                        private Long maxOutOfOrderness = 5000L;
                        @Override
                        public Watermark getCurrentWatermark() { 
                            return new Watermark(currentTimeStamp - maxOutOfOrderness);
                        }
                        @Override
                        public long extractTimestamp(String s, long l) { 
                            String[] arr = s.split(",");
                            long timeStamp = Long.parseLong(arr[1]);
                            currentTimeStamp = Math.max(timeStamp, currentTimeStamp);
                            System.err.println(s + ",EventTime:" + timeStamp + ",watermark:" + (currentTimeStamp - maxOutOfOrderness));
                            return timeStamp;
                        }
                    });
            dataStream.map(new MapFunction<String, Tuple2<String, Long>>() { 
                @Override
                public Tuple2<String, Long> map(String s) throws Exception { 
                    String[] split = s.split(",");
                    return new Tuple2<String, Long>(split[0], Long.parseLong(split[1]));
                }
            })
                    .keyBy(0)
                    .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                    .aggregate(new AggregateFunction<Tuple2<String,Long>, Object, Object>() { 
                        ...
                    })
                    .print();
            env.execute("WaterMark Test Demo");
        }//
    }

在这个案例中，我们使用的 AssignerWithPeriodicWatermarks 来自定义水印发射器和时间戳提取器，设置允许乱序时间为 5 秒，并且在一个 5 秒的窗口内进行聚合计算。  
在这个案例中，可以看到如何正确使用 Flink 提供的 API 进行水印和时间戳的设置。

# 第二十六讲 Flink中的聚合函数和累加器的设计和使用 #

我们在第 08 课时中提到了 Flink 所支持的窗口和时间类型，并且在第 25 课时中详细讲解了如何设置时间戳提取器和水印发射器。

实际的业务中，我们在使用窗口的过程中一定是基于窗口进行的聚合计算。例如，计算窗口内的 UV、PV 等，那么 Flink 支持哪些基于窗口的聚合函数？累加器又该如何实现呢？

**Flink 支持的窗口函数**

我们在定义完窗口以后，需要指定窗口上进行的计算。目前 Flink 支持的窗口函数包含 3 种：

*  ReduceFunction 增量聚合
 *  AggregateFunction 增量聚合
 *  ProcessWindowFunction 全量聚合

最后还有一种 FlodFunction，但是在 Flink 1.9 版本后已经废弃，推荐使用 AggregateFunction 代替。

下面我们详细讲解以上 3 种窗口聚合函数的定义和使用。

**ReduceFunction**

ReduceFunction 基于两个类型一致的输入进行增量聚合，我们可以自定义 ReduceFunction 来增量聚合窗口内的数据。

可以这样定义自己的 ReduceFunction，覆写 reduce 方法：

DataStream<Tuple2<String, Long>> input = ...;
    input
        .keyBy(<key selector>)
        .window(<window assigner>)
        .reduce(new ReduceFunction<Tuple2<String, Long>> { 
          public Tuple2<String, Long> reduce(Tuple2<String, Long> v1, Tuple2<String, Long> v2) { 
            return new Tuple2<>(v1.f0, v1.f1 + v2.f1);
          }
        });

其中，我们覆写的 reduce 函数接受两个参数 v1 和 v2，这两个入参的类型一致。本例中返回的是入参的第二个参数的和。

下面举个例子，我们需要计算班级中每个学生的总成绩：

public class MyReduceFunction { 
        public static void main(String[] args) throws Exception { 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            DataStream<Tuple2<String, Integer>> input = env.fromElements(courses);
            DataStream<Tuple2<String, Integer>> total = input.keyBy(0).reduce(new ReduceFunction<Tuple2<String, Integer>>() { 
                @Override
                public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception { 
                    return new Tuple2<>(value1.f0, value1.f1 + value2.f1);
                }
            });
            total.printToErr();
            env.execute("ReduceFunction");
        }
        public static final Tuple2[] courses = new Tuple2[]{ 
                Tuple2.of("张三",100),
                Tuple2.of("李四",80),
                Tuple2.of("张三",80),
                Tuple2.of("李四",95),
                Tuple2.of("张三",90),
                Tuple2.of("李四",100),
        };
    }

在上面的案例中，将输入成绩按照学生姓名进行分组，然后自定义 ReduceFunction，覆写了其中的 reduce 函数，函数中将成绩进行相加，最后打印输出。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 25]  
**AggregateFunction**

AggregateFunction 是 Flink 提供的一个通用的聚合函数实现，用户定义的聚合函数可以通过扩展 AggregateFunction 类来实现。AggregateFunction 更加通用，它有 3 个参数：输入类型（IN）、累加器类型（ACC）和输出类型（OUT）。

我们通过实现 AggregateFunction 接口，覆写下面几个方法：

public interface AggregateFunction<IN, ACC, OUT> extends Function, Serializable { 
       ACC createAccumulator();
       ACC add(IN value, ACC accumulator);
       OUT getResult(ACC accumulator);
       ACC merge(ACC a, ACC b);
    }

*  createAccumulator()：用来创建一个累加器，负责将输入的数据进行迭代
 *  add()：该函数是用来将输入的每条数据和累加器进行计算的具体实现
 *  getResult()：从累加器中获取计算结果
 *  merge()：将两个累加器进行合并

我们举个例子，自定义一个 AverageAggregate 实现 AggregateFunction 接口：

private static class AverageAggregate
        implements AggregateFunction<Tuple2<String, Long>, Tuple2<Long, Long>, Double> { 
      @Override
      public Tuple2<Long, Long> createAccumulator() { 
        return new Tuple2<>(0L, 0L);
      }
      @Override
      public Tuple2<Long, Long> add(Tuple2<String, Long> value, Tuple2<Long, Long> accumulator) { 
        return new Tuple2<>(accumulator.f0 + value.f1, accumulator.f1 + 1L);
      }
      @Override
      public Double getResult(Tuple2<Long, Long> accumulator) { 
        return ((double) accumulator.f0) / accumulator.f1;
      }
      @Override
      public Tuple2<Long, Long> merge(Tuple2<Long, Long> a, Tuple2<Long, Long> b) { 
        return new Tuple2<>(a.f0 + b.f0, a.f1 + b.f1);
      }
    }

在上面的案例中，我们自定义的 AverageAggregate 用来计算输入数据第二个字段的平均值。

**ProcessWindowFunction**

ProcessWindowFunction 用来进行全量聚合，窗口中需要维护全部原始数据，当窗口触发计算时，则进行全量聚合。ProcessWindowFunction 中有一个比较重要的对象，那就是 Context，可以用来访问事件和状态信息。但 ProcessWindowFunction 中的数据不是增量聚合，所以会使得资源消耗变大。

我们在自定义 ProcessWindowFunction 时可以实现的函数如下：

public abstract class ProcessWindowFunction<IN, OUT, KEY, W extends Window> extends AbstractRichFunction { 
       private static final long serialVersionUID = 1L;
       public abstract void process(KEY key, Context context, Iterable<IN> elements, Collector<OUT> out) throws Exception;
       public void clear(Context context) throws Exception { }
       public abstract class Context implements java.io.Serializable { 
          public abstract W window();
          public abstract long currentProcessingTime();
          public abstract long currentWatermark();
          public abstract KeyedStateStore windowState();
          public abstract KeyedStateStore globalState();
          public abstract <X> void output(OutputTag<X> outputTag, X value);
       }
    }

举个例子，实现针对窗口的分组统计功能：

DataStream<Tuple2<String, Long>> input = ...;
    input
      .keyBy(t -> t.f0)
      .timeWindow(Time.minutes(5))
      .process(new MyProcessWindowFunction());
      
    public class MyProcessWindowFunction
        extends ProcessWindowFunction<Tuple2<String, Long>, String, String, TimeWindow> { 
      @Override
      public void process(String key, Context context, Iterable<Tuple2<String, Long>> input, Collector<String> out) { 
        long count = 0;
        for (Tuple2<String, Long> in: input) { 
          count++;
        }
        out.collect("Window: " + context.window() + "count: " + count);
      }
    }

除了上述的用法，ProcessWindowFunction 还可以结合 ReduceFunction、AggregateFunction，或者 FoldFunction 来做增量计算。

例如，下面的示例是将 ReduceFunction 和 ProcessWindowFunction 结合使用返回窗口中的最小事件以及窗口的开始时间。

DataStream<SensorReading> input = ...;
    input
      .keyBy(<key selector>)
      .timeWindow(<duration>)
      .reduce(new MyReduceFunction(), new MyProcessWindowFunction());
    private static class MyReduceFunction implements ReduceFunction<SensorReading> { 
      public SensorReading reduce(SensorReading r1, SensorReading r2) { 
          return r1.value() > r2.value() ? r2 : r1;
      }
    }
    private static class MyProcessWindowFunction
        extends ProcessWindowFunction<SensorReading, Tuple2<Long, SensorReading>, String, TimeWindow> { 
      public void process(String key,
                        Context context,
                        Iterable<SensorReading> minReadings,
                        Collector<Tuple2<Long, SensorReading>> out) { 
          SensorReading min = minReadings.iterator().next();
          out.collect(new Tuple2<Long, SensorReading>(window.getStart(), min));
      }
    }

**Flink 中累加器的使用**

Accumulator（累加器）主要用来获取不同并行实例中全局指标值。如果没有累加器，我们只能获取单个实例的统计信息和聚合值。Flink 提供了累加器用于作业结束后统计全局信息。

Flink 提供了以下几类累加器供我们使用：

*  IntCounter
 *  LongCounter
 *  DoubleCounter
 *  自定义实现 Accumulator 或 SimpleAccumulator 接口

接下来我们看一下累加器的具体使用案例。

首先需要创建累加器，然后给累加器进行命名，这里需要注意，累加器的名字要全局唯一，接着将累加器注册到 Flink 的上下文中：

dataStream.map(new RichMapFunction<String, String>() { 
        //第一步：定义累加器
        private IntCounter intCounter = new IntCounter();
        @Override
        public void open(Configuration parameters) throws Exception { 
            super.open(parameters);
            //第二步：注册累加器
            getRuntimeContext().addAccumulator("counter", this.intCounter);
        }
        @Override
        public String map(String s) throws Exception { 
            //第三步：累加
            this.intCounter.add(1);
            return s;
        }
    });

最后当 Flink 程序执行完成后，可以获取累加器的值：

Object counte = jobExecutionResult.getAccumulatorResult("counter");

如果你不需要将最终的值进行持久化存储，那么可以不用获取该结果。因为该结果可以在 Flink UI 中看到。

完整的案例如下，我们计算 9000 端口中输入数据的个数：

public class CounterTest { 
        public static void main(String[] args) throws Exception { 
            final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            DataStream<String> dataStream = env.socketTextStream("127.0.0.1", 9000, "\n");
            dataStream.map(new RichMapFunction<String, String>() { 
                //定义累加器
                private IntCounter intCounter = new IntCounter();
                @Override
                public void open(Configuration parameters) throws Exception { 
                    super.open(parameters);
                    //注册累加器
                    getRuntimeContext().addAccumulator("counter", this.intCounter);
                }
                @Override
                public String map(String s) throws Exception { 
                    //累加
                    this.intCounter.add(1);
                    return s;
                }
            });
            dataStream.print();
            JobExecutionResult result = env.execute("counter");
            //第四步：结束后输出总量；如果不需要结束后持久化，可以省去
            Object accResult = result.getAccumulatorResult("counter");
            System.out.println("累加器计算结果:" + accResult);
        }
    }

我们在 9000 端口中输入几行数据：  
![在这里插入图片描述][20200927202440680.png_pic_center]  
然后断开端口，可以在控制台中看到输出的结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 26]

# 第二十七讲 Flink Redis Sink实现 #

我们在第 12 课时“Flink 常用的 Source 和 Connector”中提过 Flink 提供了比较丰富的用来连接第三方的连接器，可以在[官网][Link 7]中找到 Flink 支持的各种各样的连接器。

此外，Flink 还会基于 Apache Bahir 发布一些 Connector，其中就有我们非常熟悉的 Redis。很多人在 Flink 项目中访问 Redis 的方法都是自己进行实现的，我们也可以使用 Bahir 实现的 Redis 连接器。

事实上，使用 Redis Sink 常用的方法有很多，比如自定义 Sink 函数、依赖开源的 Redis Sink 实现等。

下面我们就分别介绍常用的 Redis Sink 实现。

**自定义 Redis Sink**

> REmote DIctionary Server（Redis）是一个由 Salvatore Sanfilippo 写的 key-value 存储系统。Redis 是一个开源的使用 ANSI C 语言编写、遵守 BSD 协议、支持网络、可基于内存亦可持久化的日志型、Key-Value 数据库，并提供多种语言的 API。

> 它通常被称为数据结构服务器，因为值（value）可以是字符串（String）、哈希（Hash）、列表（List）、集合（Sets）和有序集合（Sorted Sets）等类型。

如果你对 Redis 不熟悉，可以参考[官网][Link 8]上的说明。 点击[这里][Link 9]下载一个稳定版本的 Redis，我在本地安装的是 2.8.5 版本。使用下面命令进行安装：

wget http://download.redis.io/releases/redis-2.8.5.tar.gz
    tar xzf redis-2.8.5.tar.gz
    cd redis-2.8.5
    make

然后进入 src 目录，就可以在本地启动一个 Redis 单机实例。

src/redis-server

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 27]  
我们可以使用 Redis 自带的交互命令来测试 Redis 实例，如下图所示：  
![在这里插入图片描述][20200928102700530.png_pic_center]  
从上图可以看到，我们通过命令可以向 Redis 中设置数据。

然后在使用 Redis 时需要新增新的依赖，你可以点击[这里][Link 10]根据 Redis 版本找到对应的 Redis 依赖：

<dependency>
       <groupId>redis.clients</groupId>
       <artifactId>jedis</artifactId>
       <version>2.8.2</version>
    </dependency>

我们自定义一个 RedisSink 函数，继承 RichSinkFunction，重写其中的方法：

public class SelfRedisSink extends RichSinkFunction { 
        private transient Jedis jedis;
        public void open(Configuration config) { 
            jedis = new Jedis("localhost", 6379);
        }
        public void invoke(Tuple2<String, String> value, Context context) throws Exception { 
            if (!jedis.isConnected()) { 
                jedis.connect();
            }
            jedis.set(value.f0, value.f1);
        }
        @Override
        public void close() throws Exception { 
            jedis.close();
        }
    }

上述代码，我们继承了 RichSinkFunction 并且实现了其中的 open、invoke 和 close 方法。其中 open 用于新建 Redis 客户端；invoke 函数将数据存储到 Redis 中，我们在这里将数据以字符串的形式存储到 Redis 中；close 函数用于使用完毕后关闭 Redis 客户端。

我们使用自定义 Redis Sink 的优点在于可以根据业务需求灵活处理，可以方便地使用 Jedis 提供的各种能力。

**使用开源 Redis Connector**

常用的开源 Connector 有两个，分别是 Flink 和 Bahir 提供的实现，其内部都是使用了 Java Redis 客户端 Jedis 实现了 Redis Sink。我们分别看一下它们的使用方法。

**第一种：Flink 提供的依赖包**

新增以下依赖：

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

我们可以通过实现 RedisMapper 来自定义 Redis Sink：

public class RedisSink implements RedisMapper<Tuple2<String, String>>{ 
        /** * 设置 Redis 数据类型 */
        @Override
        public RedisCommandDescription getCommandDescription() { 
            return new RedisCommandDescription(RedisCommand.SET);
        }
        /** * 设置Key */
        @Override
        public String getKeyFromData(Tuple2<String, String> data) { 
            return data.f0;
        }
        /** * 设置value */
        @Override
        public String getValueFromData(Tuple2<String, String> data) { 
            return data.f1;
        }
    }

我们自定义的 RedisSink 实现了 RedisMapper 并覆写了其中的 getCommandDescription、getKeyFromData、getValueFromData。其中 getCommandDescription 定义了存储到 Redis 中的数据格式，这里我们定义的 RedisCommand 为 SET，将数据以字符串的形式存储；getKeyFromData 定义了 SET 的 Key，getValueFromData 定义了 SET 的值。

完整的使用方式为：

public static void main(String[] args) throws Exception{ 
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<Tuple2<String, String>> stream = env.fromElements("Flink","Spark","Storm").map(new MapFunction<String, Tuple2<String, String>>() { 
            @Override
            public Tuple2<String, String> map(String s) throws Exception { 
                return new Tuple2<>(s, s);
            }
        });
        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("localhost").setPort(6379).build();
        stream.addSink(new RedisSink<>(conf, new RedisSink01()));
        env.execute("redis sink01");
    }

我们可以在 Redis 的控制台中查询数据，如下图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 28]  
由图可知，我们设置了 Redis 中的数据。

**第二种：Bahir 提供的依赖包**

新增以下依赖：

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

目前该开源的 Bahir 实现最新版本是 1.1-snapshot，在这里使用的是 1.0 稳定版本。同样，我们也可以通过实现 RedisMapper 来自定义 Redis Sink：

public class RedisSink02 implements RedisMapper<Tuple2<String, String>> { 
        /** * 设置 Redis 数据类型 */
        @Override
        public RedisCommandDescription getCommandDescription() { 
            return new RedisCommandDescription(RedisCommand.SET);
        }
        /** * 设置 Key */
        @Override
        public String getKeyFromData(Tuple2<String, String> data) { 
            return data.f0;
        }
        /** * 设置 value */
        @Override
        public String getValueFromData(Tuple2<String, String> data) { 
            return data.f1;
        }
    }

和第一种方式一样，完整的使用方式如下代码所示：

public static void main(String[] args) throws Exception{ 
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<Tuple2<String, String>> stream = env.fromElements("Flink","Spark","Storm").map(new MapFunction<String, Tuple2<String, String>>() { 
            @Override
            public Tuple2<String, String> map(String s) throws Exception { 
                return new Tuple2<>(s, s+"_sink2");
            }
        });
        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("localhost").setPort(6379).build();
        stream.addSink(new RedisSink<>(conf, new RedisSink02()));
        env.execute("redis sink02");
    }

我们可以在 Redis 的控制台中查询数据，如下图所示：  
![在这里插入图片描述][20200928103915220.png_pic_center]  
开源实现的 Redis Connector 使用非常方便，但是有些功能缺失，例如，无法使用一些 Jedis 中的高级功能如设置过期时间等。

所以我们在实际生产中可以根据自己业务需要使用不同的实现。

# 第二十八讲 TopN 热门商品功能实现 #

本课时主要讲解 Flink 中的 TopN 功能的设计和实现。

TopN 在我们的业务场景中是十分常见的需求，比如电商场景中求热门商品的销售额、微博每天的热门话题 TopN、贴吧中每天发帖最多的贴吧排名等。TopN 可以进行分组排序，也可以按照需要全局排序，比如若要计算用户下单总金额的 Top 10 时，就需要进行全局排序，然而当我们计算每个城市的 Top10 时就需要将订单按照城市进行分组然后再进行计算。

下面我们就详细讲解 TopN 的设计和实现。

**整体设计**

我们下面使用订单数据进行讲解，整体的数据流向如下图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 29]  
订单数据由业务系统产生并发送到 Kafka 中，我们的 Flink 代码会消费 Kafka 中的数据，并进行计算后写入 Redis，然后前端就可以通过读取 Redis 中的数据进行展示了。

**订单设计**

简化后的订单数据如下，主要包含：下单用户 ID、商品 ID、用户所在城市名称、订单金额和下单时间。

public class OrderDetail { 
        private Long userId; //下单用户id
        private Long itemId; //商品id
        private String citeName;//用户所在城市
        private Double price;//订单金额
        private Long timeStamp;//下单时间
    }

我们采用 Event-Time 来作为 Flink 程序的时间特征，并且设置 Checkpoint 时间周期为 60 秒。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    env.enableCheckpointing(60 * 1000, CheckpointingMode.EXACTLY_ONCE);
    env.getCheckpointConfig().setCheckpointTimeout(30 * 1000);

**Kafka Consumer 实现**

我们在第 24 课时“Flink 消费 Kafka 数据业务开发” 中详细讲解过 Kafka Consumer 的实现，在这里订阅 Kafka 消息作为数据源，设置从最早位点开始消费数据：

Properties properties = new Properties();
    properties.setProperty("bootstrap.servers", "localhost:9092");
    FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("test", new SimpleStringSchema(), properties);
    //从最早开始消费
    consumer.setStartFromEarliest();
    DataStream<String> stream = env
            .addSource(consumer);

**时间提取和水印设置**

因为订单消息流可能存在乱序的问题，我们在这里设置允许乱序时间为 30 秒，并且设置周期性水印：

DataStream<OrderDetail> orderStream = stream.map(message -> JSON.parseObject(message, OrderDetail.class));
    orderStream.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<OrderDetail>() { 
        private Long currentTimeStamp = 0L;
        //设置允许乱序时间
        private Long maxOutOfOrderness = 5000L;
        @Override
        public Watermark getCurrentWatermark() { 
            return new Watermark(currentTimeStamp - maxOutOfOrderness);
        }
        @Override
        public long extractTimestamp(OrderDetail element, long previousElementTimestamp) { 
            return element.getTimeStamp();
        }
    });

**下单金额 TopN**

我们在这里要求所有用户中下单金额最多的 Top 10 用户，这里就会用到 windowAll 函数：

DataStream<OrderDetail> reduce = dataStream
            .keyBy((KeySelector<OrderDetail, Object>) value -> value.getUserId())
            .windowAll(SlidingProcessingTimeWindows.of(Time.seconds(600), Time.seconds(20)))
            .reduce(new ReduceFunction<OrderDetail>() { 
                @Override
                public OrderDetail reduce(OrderDetail value1, OrderDetail value2) throws Exception { 
                    return new OrderDetail(
                            value1.getUserId(), value1.getItemId(), value1.getCiteName(), value1.getPrice() + value2.getPrice(), value1.getTimeStamp()
                    );
                }
            });
    
    //每20秒计算一次
    SingleOutputStreamOperator<Tuple2<Double, OrderDetail>> process = reduce.windowAll(TumblingEventTimeWindows.of(Time.seconds(20)))
            .process(new ProcessAllWindowFunction<OrderDetail, Tuple2<Double, OrderDetail>, TimeWindow>() { 
                         @Override
                         public void process(Context context, Iterable<OrderDetail> elements, Collector<Tuple2<Double, OrderDetail>> out) throws Exception { 
                             TreeMap<Double, OrderDetail> treeMap = new TreeMap<Double, OrderDetail>(new Comparator<Double>() { 
                                 @Override
                                 public int compare(Double x, Double y) { 
                                     return (x < y) ? -1 : 1;
                                 }
                             });
                             Iterator<OrderDetail> iterator = elements.iterator();
                             if (iterator.hasNext()) { 
                                 treeMap.put(iterator.next().getPrice(), iterator.next());
                                 if (treeMap.size() > 10) { 
                                     treeMap.pollLastEntry();
                                 }
                             }
                             for (Map.Entry<Double, OrderDetail> entry : treeMap.entrySet()) { 
                                 out.collect(Tuple2.of(entry.getKey(), entry.getValue()));
                             }
                         }
                     }
            );

这段代码实现略显复杂，核心逻辑如下：  
首先对输入流按照用户 ID 进行分组，并且自定义了一个滑动窗口，即 SlidingProcessingTimeWindows.of(Time.seconds(600), Time.seconds(20))，表示定义一个总时间长度为 600 秒，每 20 秒向后滑动一次的滑动窗口。

经过上面的处理后，我们的订单数据会按照用户维度每隔 20 秒进行一次计算，并且通过 windowAll 函数将所有的数据汇聚到一个窗口。

这里需要注意，windowAll 是一个并发度为 1 的特殊操作，也就是所有元素都会进入到一个窗口内进行计算。

那么我们是如何取得 Top 10 的呢？

在这里，我们定义了一个 TreeMap，TreeMap 存储 K-V 键值对，通过红黑树（R-B tree）实现，红黑树结构天然支持排序，默认情况下通过 Key 值的自然顺序进行排序。我们设置的 TreeMap 大小是 10，如果新的数据到来后，TreeMap 的数据已经到达 10个，那么就会进行比较，将较小的删除。

**写入 Redis**

我们在第 27 课时中讲了 Flink Redis Sink 的实现，把结果存入 Redis 中，结构为 HASH：

FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("localhost").setPort(6379).build();
    process.addSink(new RedisSink<>(conf, new RedisMapper<Tuple2<Double, OrderDetail>>() { 
        private final String TOPN_PREFIX = "TOPN:";
        @Override
        public RedisCommandDescription getCommandDescription() { 
            return new RedisCommandDescription(RedisCommand.HSET,TOPN_PREFIX);
        }
        @Override
        public String getKeyFromData(Tuple2<Double, OrderDetail> data) { 
            return String.valueOf(data.f0);
        }
        @Override
        public String getValueFromData(Tuple2<Double, OrderDetail> data) { 
            return String.valueOf(data.f1.toString());
        }
    }));

我们的 RedisSink 通过新建 RedisMapper 来实现，覆写了其中的 getCommandDescription、getKeyFromData、getValueFromData 三个方法。分别用来设定存储数据的格式，Redis 中 Key 的实现和 Value 的实现。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/7e38e333454441ac947ab0316e92fbb8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/ebd010b5610b484bb4672e05473c0128.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/5b840430c43047e6b982bace9f4f1566.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221123/63024023ca9e4d39b6c5f0cf9fff4b83.png
[Flink]: https://ci.apache.org/projects/flink/flink-docs-release-1.10/monitoring/metrics.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221123/e1d8ca57d203432a818f0e3f341154e2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221123/d4844e59a3cd4cd8aef6b3f47d2d8301.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 6]: /images/20221123/1fb54911c0834bdf84284b470e979f7d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 7]: /images/20221123/4d82292eb1cb4721960d5903599fa71a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 8]: /images/20221123/e20bbd3517104d5ead3f40ab77a05faa.png
[20200917110704465.png_pic_center]: /images/20221123/b64d8eedb24f42a1bdf942ca01af05f8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 9]: /images/20221123/1bf07566929c45078fb68dfd0bfbd148.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 10]: /images/20221123/6fd3cafbfe1a4f59bd50a59626770996.png
[Link 1]: https://ci.apache.org/projects/flink/flink-docs-release-1.10/monitoring/metrics.html#reporter
[Link 2]: https://grafana.com/docs/grafana/latest/installation/rpm/
[Link 3]: https://mp.weixin.qq.com/s/9MZ9ztr8fYJTl1HchqtQqA
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 11]: /images/20221123/a382eefcda0a47fdb13a8429f307a96c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 12]: /images/20221123/d0f5113e034f4f2abbf30ac166013fbe.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 13]: /images/20221123/5c0ea3051c4344a095f5f20d1ce8c856.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 14]: /images/20221123/83150f80fa684de6adc50eb11b403d3a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 15]: /images/20221123/2120604671a440209e10029a536f2d34.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 16]: /images/20221123/b3c367782f7f41f3af72f5ba3fa91d48.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 17]: /images/20221123/8025c44592db4e93a519b9cd944e4c17.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 18]: /images/20221123/54a1c84f22e9430886db60af4c11ec9b.png
[Link 4]: https://github.com/alibaba/canal
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 19]: /images/20221123/5148008e9dd4473ebddc1a9e6ad141bd.png
[Link 5]: https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/kafka.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 20]: /images/20221123/58c5578caa5049968438de6b2fa23468.png
[Link 6]: http://kafka.apache.org/downloads
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 21]: /images/20221123/1ebdffabc2d044b7b641474ce3d6570f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 22]: /images/20221123/b886095203b7437394d8c8bc6f04a5bb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 23]: /images/20221123/d9cab73a3dfb4de297872871a89d6598.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 24]: /images/20221123/22d826ec312149c49d8ad1e8767ecdf9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 25]: /images/20221123/6ebf1b0de710436694771efdc3973033.png
[20200927202440680.png_pic_center]: /images/20221123/f0f092b784874f9caed91e2088a3a4eb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 26]: /images/20221123/321244bd97334d8d841e73453dc035c0.png
[Link 7]: https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/connectors/
[Link 8]: https://redis.io/
[Link 9]: http://download.redis.io/releases/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 27]: /images/20221123/112b945ab9d14bdba4b4ce037d4b1a04.png
[20200928102700530.png_pic_center]: /images/20221123/4d36bd5cd75f4875bfc65dac136dd2ff.png
[Link 10]: https://mvnrepository.com/artifact/redis.clients/jedis
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 28]: /images/20221123/59c3151e65bf4a3f82cd6fafe178481f.png
[20200928103915220.png_pic_center]: /images/20221123/de83b331132241cd90dfd96c7c9b52f1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MDk1MDU1_size_16_color_FFFFFF_t_70_pic_center 29]: /images/20221123/1f4de4df602c416eb9afe9f043c7a2cb.png