flink-流上的join (1.12.2)

我就是我 2022-10-13 05:04 142阅读 0赞

# flink-流上的join (1.12.2) #

### 一、简介 ###

本篇主要围绕flink table & sql进行实践。

其中文章中大部分内容官方文档有详细记载。

由于环境和代码编写问题，官方示例并不能完全用于代码实践，所以此篇文章通过kafka+mysql进行简单实践。

环境：

*  kafka 2.4
 *  flink 1.12.2
 *  mysql57
 *  java8

### 二、时间属性 ###

首先对于实时join来说，第一个要讨论的就是时间属性。

在flink sql中，给出了两种时间属性，一个是系统处理时间，一个是数据本身的自带的时间属性。

对于处理时间来说，大部分针对普通的join，对数据顺序先后没有太大要求。

但是在数据自带的时间戳上，哪条数据是在哪个时间段处理便比较重要。

#### 2.1 处理时间 ####

处理时间是基于机器的本地时间来处理数据，它是最简单的一种时间概念，但是它不能提供确定性。它既不需要从数据里获取时间，也不需要生成 watermark。

bsTableEnv.executeSql("create table kafka_table( " +
                    "`name` STRING, " +
                    "`id` INT," +
                    "user_action_time AS PROCTIME() " +
                    ") WITH ( " +
                    "'connector.type'='kafka', " +
                    "'connector.version'='universal', " +
                    "'connector.topic'='test', " +
                    "'connector.properties.bootstrap.servers'='localhost:9092', " +
                    "'connector.properties.group.id'='flink', " +
                    "'connector.startup-mode'='earliest-offset', " +
                    "'format.type'='json' " +
                    ")");

#### 2.2 时间事件 ####

事件时间允许程序按照数据中包含的时间来处理，这样可以在有乱序或者晚到的数据的情况下产生一致的处理结果。它可以保证从外部存储读取数据后产生可以复现（replayable）的结果。

除此之外，事件时间可以让程序在流式和批式作业中使用同样的语法。在流式程序中的事件时间属性，在批式程序中就是一个正常的时间字段。

为了能够处理乱序的事件，并且区分正常到达和晚到的事件，Flink 需要从事件中获取事件时间并且产生 watermark（[watermarks][]）。

bsTableEnv.executeSql("create table kafka_table( " +
                    "`receive_time` BIGINT, " +
                    "`id` STRING," +
                    "`name` STRING," +
                    "`ts` AS TO_TIMESTAMP(FROM_UNIXTIME(`receive_time` / 1000, 'yyyy-MM-dd HH:mm:ss'))," +
                    "WATERMARK FOR ts AS ts - INTERVAL '5' SECOND " +
                    ") WITH ( " +
                    "'connector.type'='kafka', " +
                    "'connector.version'='universal', " +
                    "'connector.topic'='watermarkjoin', " +
                    "'connector.properties.bootstrap.servers'='localhost:9092', " +
                    "'connector.properties.group.id'='flink', " +
                    "'connector.startup-mode'='earliest-offset', " +
                    "'format.type'='json' " +
                    ")");

**注意**

> 在 Table 编程环境中，基于 SQL 的类型系统与程序指定的数据类型之间需要物理提示。该提示指出了实现预期的数据格式。
> 
> 例如，Data Source 能够使用类 `java.sql.Timestamp` 来表达逻辑上的 `TIMESTAMP` 产生的值，而不是使用缺省的 `java.time.LocalDateTime`。有了这些信息，运行时就能够将产生的类转换为其内部数据格式。反过来，Data Sink 可以声明它从运行时消费的数据格式。

当使用时间事件时，取值为数据自带的时间，一般为xxxx-xx-xx xx:xx:xx，但是在flink sql中进行转换时：

`receive_time` TIMESTAMP(3)

会出现以下错误（时间戳13位）：

Caused by: java.time.format.DateTimeParseException: Text '1618987432830' could not be parsed at index 0

或者（yyyy-MM-dd HH:mm:ss时间格式）

Caused by: java.time.format.DateTimeParseException: Text '2021-04-21 11:11:23' could not be parsed at index 10

所以此处用

`receive_time` BIGINT,
    `ts` AS TO_TIMESTAMP(FROM_UNIXTIME(`receive_time` / 1000, 'yyyy-MM-dd HH:mm:ss'))

进行了转换，并进行了查询发现其时间类型为`2021-04-22T00:12:30`才能进行转换。

Kafka Producer生产者生成消息：

Integer i = random.nextInt(10);
                Timestamp timestamp = new Timestamp(System.currentTimeMillis());
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("id", i.toString());
                jsonObject.put("name", "test" + i);
                jsonObject.put("receive_time", timestamp.getTime());
    
    ProducerRecord<String, String> record = new ProducerRecord<>("watermarkjoin", i.toString(),
                                                                 jsonObject.toString());

以30s内滚动窗口进行查询，30s进行一次统计：

TableResult result = bsTableEnv.executeSql("select TUMBLE_START(ts, INTERVAL '30' SECOND), COUNT(DISTINCT id) from kafka_table group by " +
                    "TUMBLE(ts, INTERVAL '30' SECOND)");
    
    result.print();

结果为：

±—±------------------------±---------------------+  
| op | EXPR$0 | EXPR$1 |  
±—±------------------------±---------------------+  
| +I | 2021-04-21T14:43:30 | 7 |  
| +I | 2021-04-21T14:44 | 10 |  
| +I | 2021-04-21T14:44:30 | 10 |  
| +I | 2021-04-21T14:45 | 10 |  
| +I | 2021-04-21T14:45:30 | 10 |  
| +I | 2021-04-21T14:46 | 10 |  
| +I | 2021-04-21T14:46:30 | 10 |

### 三、流上join ###

#### JOIN 当前维表 ####

Flink SQL 支持 LEFT JOIN 和 INNER JOIN 的维表关联。语法所示，维表 JOIN 语法与传统的 JOIN 语法并无二异。只是 Products 维表后面需要跟上 `FOR SYSTEM_TIME AS OF PROCTIME()` 的关键字，其含义是每条到达的数据所关联上的是到达时刻的维表快照，也就是说，当数据到达时，我们会根据数据上的 key 去查询远程数据库，拿到匹配的结果后关联输出。这里的 `PROCTIME` 即 processing time。使用 JOIN 当前维表功能需要注意的是，如果维表插入了一条数据能匹配上之前左表的数据时，JOIN的结果流，不会发出更新的数据以弥补之前的未匹配。JOIN行为只发生在处理时间（processing time），即使维表中的数据都被删了，之前JOIN流已经发出的关联上的数据也不会被撤回或改变。

SELECT *
    FROM Orders AS o
    [LEFT] JOIN Products FOR SYSTEM_TIME AS OF PROCTIME() AS p
    ON o.productId = p.productId

#### JOIN 历史维表 ####

有时候想关联上的维度数据，并不是当前时刻的值，而是某个历史时刻的值。比如，产品的价格一直在发生变化，订单流希望补全的是下单时的价格，而不是当前的价格，那就是 JOIN 历史维表。语法上只需要将上文的 `PROCTIME()` 改成 `o.orderTime` 即可。含义是关联上的是下单时刻的 Products 维表。

SELECT *
    FROM Orders AS o
    [LEFT] JOIN Products FOR SYSTEM_TIME AS OF o.orderTime AS p
    ON o.productId = p.productId

Flink 在获取维度数据时，会根据左流的时间去查对应时刻的快照数据。因此 JOIN 历史维表需要外部存储支持多版本存储，如 HBase，或者存储的数据中带有多版本信息。

#### 多表JOIN ####

INSERT INTO sink_print
    SELECT assure_orders.order_id,assure_orders.order_name,dim_base_province.name FROM assure_orders
    LEFT JOIN dim_base_province FOR SYSTEM_TIME AS OF assure_orders.proctime
    ON assure_orders.order_id = dim_base_province.id

INSERT INTO sink_print
     select d.order_id,d.company_name,d.address_name,d.order_name,dim_base_province.name from
     (
     SELECT a.*,b.*,c.*
     FROM assure_orders a
     JOIN
     order_company b
      ON a.order_id = b.company_id
     JOIN
     order_address c
     ON a.order_id = c.address_id
      ) d
     LEFT JOIN
     dim_base_province FOR SYSTEM_TIME AS OF d.proctime
     ON d.order_id = dim_base_province.id

在多表join的时候，如果都是全量join，那必定会出现state的增长，最终导致OOM。

所以一般我们想到的就是设置state的过期时间

bsTableEnv.getConfig.setIdleStateRetentionTime(Time.seconds(30), Time.minutes(6))

#### 维表优化 ####

针对维表关联的情况，为了降低 IO 请求次数，降低维表数据库读压力，从而降低延迟，提高吞吐，有以下几种措施：

*  当维表数据量不大时，通过全量维表数据缓存在本地，同时 ttl 控制缓存刷新的时候，这可以极大的降低 IO 请求次数，但会要求更多的内存空间。
 *  当维表数据量很大时，通过 async 和 LRU cache 策略，同时 ttl 和 size 来控制缓存数据的失效时间和缓存大小，可以提高吞吐率并降低数据库的读压力。
 *  当维表数据量很大同时主流 qps 很高时，可以开启把维表 join 的 key 作为 hash 的条件，将数据进行分区，即在 calc 节点的分区策略是 hash，这样下游算子的 subtask 的维表数据是独立的，不仅可以提高命中率，也可降低内存使用空间。

> 来自唯品会的flink实践

![weipinghui flink sql][]

#### **维表关联延迟 join** ####

维表关联中，有很多业务场景，在维表数据新增数据之前，主流数据已经发生 join 操作，会出现关联不上的情况。因此，为了保证数据的正确，将关联不上的数据进行缓存，进行延迟 join。

*  最简单的做法是，在维表关联的 function 里设置重试次数和重试间隔，这个方法会增大整个流的延迟，但主流 qps 不高的情况下，可以解决问题。
 *  增加延迟 join 的算子，当 join 维表未关联时，先缓存起来，根据设置重试次数和重试间隔从而进行延迟的 join。

[watermarks]: https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/dev/event_time.html
[weipinghui flink sql]: /images/20221005/4396167ad9be49eb983a64010ad5a5a6.png