Pravega Flink Connector Table API 进阶功能探秘

1. 项目动机

2. 项目介绍

3. 系统架构

CREATE CATALOG pravega_catalog WITH(  'type' = 'pravega',  'default-database' = 'scope1',  'controller-uri' = 'tcp://localhost:9090',  'schema-registry-uri' = 'http://localhost:9092');

实现难点

• 第一个是 schema 和序列化的处理。Flink 在 1.12 版本之前，对于 Json 和 Avro 的数据与 Flink 表的行记录的内部抽象 RowData 转换的过程是 Flink 在 format 下的内部且 private 类。因此我们想要重用这一部分代码来保持与 Flink 对于 schema 以及序列化转换行为的一致性，就不得不复制整个的类库。因此我们就向社区提出了这样一个想法，能不能把它抽象出来，并转换为 public 的类。经过与社区的讨论和沟通，我们也开了相应的 JIRA FLINK-19098^[1]且贡献代码修复了这一问题，成功打通了在 Catalog table 转换的数据序列化的链路。
  
  
• 第二点是同时支持 Avro 和 Json 的格式。在现有的 catalog 实现中，序列化方法其实是相对固定的，这其实与 Pravega Schema Registry 设计理念里面的通用性有些不一致。可能有些用户习惯使用 Avro，而其他的有些习惯使用 Json，我们如何兼顾这两者的需求，使得他们都能享受到 catalog 带来的便利呢？我们这里在 catalog 引入了 serialization format 的选项，打通了不但可以指定 Avro 和 Json，更可以添加 Flink 官方支持的所有序列化附加配置，例如 timestamp 的序列化格式，进一步定制应用于整个 catalog 序列的方法。
  
  
• 难点三就是我们通过 Schema Registry 处理的 event 开头会有五个 byte 的 header，导致没有办法直接使用 Flink 现成的 Json 和 Avro format 进行序列化，为了使用到 Schema Registry 提供的序列化器 API，我们就需要开发自己的一套 format factory 去调用。具体到实现，我们是将 Catalog table 中与序列化，也就是 Schema Registry API 所需要的相关参数，包括 namespace、group 等信息，沿袭到 format factory 参数当中去，也就相当于如下的建表 DDL。

CREATE TABLE clicks (  ...) WITH (  'connector' = 'pravega',  ...  'format' = 'pravega-registry',  'pravega-registry.uri' = 'tcp://localhost:9092',  'pravega-registry.namespace' = 'scope1',  'pravega-registry.group-id' = 'stream1',  'pravega-registry.format' = 'Avro');

然后在实现中拿到这些信息去调用 Schema Registry 的API，再拼接上之前提到的问题的修复，我们就完成了完整的二进制数据与 Flink RowData 的一个互操作。整个链路就通了。

1. 部署方式

目前 Debezium 有三种部署方式：

• 第一种，也是业界最常用的一种利用 Kafka Connect 进行部署。之前也提到，这也是 Debezium 创始之处就支持的一种使用方式，它是将 MySQL 或者 Postgres DB 这样的传统数据库的 binlog 进行分析，通过 Kafka connect 的 Debezium source 实现导到 Apache Kafka 中。利用 Kafka 强大的生态，可以对接丰富的下游引擎，来做进一步的聚合计算和数仓数据湖的应用。
  
  
• 第二点，其实 Debezium 社区也看到了在消息队列和流存储领域的变化，慢慢地剥离 Kafka。它现在可以支持启动单独的 Debezium server 使用 source connector 连接下游的消息系统。现在已有集成的例如 Amazon Kinesis 以及 Google 的 PubSub。在 2021 年上半年，Debezium 社区发布了 1.6 版本，在这个发版中，Debezium 也正式接纳了我们 Pravega 社区的贡献。Pravega 端的 sink 也成为了 source connector 的实现之一，实现了与 Debezium 社区的牵手。
  
  
• 最后一种方式是将 Debezium 作为一个依赖库嵌入  Java 程序进行调用。比较有名的就是社区非常火热的云邪和雪尽老师负责的 Flink CDC connector。在无需长期存储或是重复利用 CDC 数据的场景中，这样轻量级的实现能移除消息队列部署运维的复杂性，同时依然保证计算的可靠性和容错性。

2. 写入方式

3. Connector 集成

Debezium 提供的是一个 upsert 流，那在转化 Flink table 的 Rowdata 抽象时，对于 insert 的 event 是一个常见的一对一的反序列化的过程。但是对于 update ，我们就需要转换成更新前和更新后状态的两个 event。这和我们对于 connector 序列化的实现默认一对一是完全违背的。

由于 Pravega 自身的序列化器接口也是一个一对一的映射，并且为了保证 Pravega 序列化器与 Flink DeserializationSchema 接口的互操作性，我们也在上面做了不少的代码支持，方便用户使用。为了支持这一新的需求，我们就不得不重构之前的反序列化链路，将原有的反序列过程从 Pravega client 端上提到 Pravega connector 内部去做，从而利用如下带有 collector 的 deserialize 方法。

default void deserialize(byte[] message, Collector<T> out) throws IOException {

同时整个代码修改的链路也需要非常小心，依然保证了原有的 Pravega 序列化器与 Flink API 转化 API 的兼容性，从而不影响线上用户的可升级性。

其次是 FLIP-107，关于 Table source 端 metadata 的支持。

Pravega Table Source 实现了 Flink 提供的 SupportsReadingMetadata 的接口，提供了这样一个支持。用户可以 from-format 的前缀来指定来自 format factory 本身的 metadata，例如 Debezium 的 ingestion timestamp，table 名称等元信息来完善和丰富表的信息。同时我们还支持了来自 Pravega 本身的 metadata，称为 EventPointer 来记录当前 event 在 stream 中的位置信息，记录这一信息也有助于用户存储这一数据，然后来完成后续的随机读取甚至建立索引的需求。整个包含了 metadata 的 create table 的建表的 DDL 如下，这些新建的列在 Flink Table 的抽象中会出现在原始数据结构的后面依次排列。

CREATE TABLE debezium_source (  // Raw Schema  id INT NOT NULL,  // Format metadata  origin_ts TIMESTAMP(3) METADATA FROM 'from_format.ingestion-timestamp' VIRTUAL,  origin_table STRING METADATA FROM 'from_format.source.table' VIRTUAL,  // Connector metadata  event_pointer BYTES METADATA VIRTUAL);

滕昱 (左) - Pravega 中国社区创始人，戴尔科技集团 OSA 软件开发总监

秦江杰 (右) - Apache Flink PMC，阿里巴巴开源大数据团队生态技术负责人

FFA 2021 直播回放 & 演讲 PDF 获取

   戳我，获取 FFA 2021 峰会资料～