1、周煜敏戴尔科技集团高级软件工程师Pravega Flink connector Table API 进阶功能探秘Deep dive into the extended Table API usage of Pravega Flink connector动态表与流表二相性动态表与流表二相性#1#1Debezium支持支持#2#2Pravega Schema registry#3#3Catalog支持支持#4#4Demo#5#5#1流表二象性流表二象性Dynamic table and stream-table duality关系代数关系代数/SQL流处理流处理关系(或表)是有界(多)元组集合流是

2、一个无限元组序列对批数据(例如关系数据库中的表)执行的查询可以访问完整的输入数据流式查询在启动时不能访问所有数据，必须“等待”数据流入批处理查询在产生固定大小的结果后终止流查询不断地根据接收到的记录更新其结果，并且始终不会结束传统关系查询与流处理比较传统关系查询与流处理比较表体现数据在某个时间点的视图流体现数据在时间维度上的变化流流-表表连续查询连续查询Update动态表动态表-流流(Changelog)RetractUpsertINSERTINSERT+DELETEUPSERT+DELETEAppend-only流-表表-流在时间维度上，修改表日志流的聚合结果成为表在时间维度上，表的动态的修

3、改记录成为流流表二象性流表二象性#2Debezium支持支持Pravega Debezium supportCDC/Debezium简介简介Change Data Capture，确定和跟踪数据库中的数据变更以采取行动基于 Kafka connect 实现，将数据库行级变更转换为事件流的分布式平台CDCDebezium的三种部署模式的三种部署模式#1将 Debezium 作为依赖库嵌入 Java 程序调用利用 Kafka Connect 部署启动 Debezium Server，利用 source connector 链接下游消息系统#2#3https:/ PravegaPravega 写客户

4、端写客户端01100000Txn segments01110110 01000110 01100000s1s201111000Regular writeTransactional write#1普通写客户端writer.writeEvent(event)事务性写客户端Transaction txn=txnWriter.beginTxn()txn.writeEvent(event)txn.flush()mit()txn.abort()#2Debezium与与Pravega 集成集成#1普通写客户端集成事务性写客户端集成#2以单独event方式顺序写入一个 debezium batch 以事务写入

5、，当batch complete 时 commit Pravega transaction，确保仅一次语义TXN-2TXN-1（)配置默认值详情debezium.sink.type设置为 pravegadebezium.sink.pravega.controller.uri tcp:/localhost:9090Pravega 集群的 controller 地址debezium.sink.pravega.scopePravega scope 名称debezium.sink.pravega.transactionfalse是否使用事务性写入模式Pravega Flink Connector 集成

6、集成基于现有 FLIP-95 Table Source API 以及 SourceFunction 实现Insert(+I,RowData)Update(-U,OldData)(+U,NewData)Issue 530:额外支持反序列化成多个 event 的功能Issue 494:FLIP-107 metadata 在 table source 上的支持为保持 Pravega 序列化器的互操作性，重构反序列化链路#3Pravega Schema RegistryWriterReadersPravega StreamReadersRaw bytesWriter“type”:“record”,“n

7、ame”:“Person”,“fields”:“name”:“age”,“type”:“int”,“name”:“name”,“type”:“string”,“name”:“gender”,“type”:“string”“type”:“record”,“name”:“Person”,“fields”:“name”:“age”,“type”:“int”,“name”:“name”,“type”:“string”Event v2Event v1“type”:“record”,“name”:“Person”,“fields”:“name”:“age”,“type”:“int”,“name”:“nam

8、e”,“type”:“string”“type”:“record”,“name”:“Person”,“fields”:“name”:“age”,“type”:“int”,“name”:“name”,“type”:“string”,“name”:“gender”,“type”:“string”项目动机项目动机存储和管理存储在 Pravega 中的半结构化数据的SchemaSchema 存储145632提供了 RESTful 接口来存储和管理数据Schema 以及兼容性策略RESTful 接口内置对 Avro、Profobuf 和 JSON 等流行的序列化格式的支持，同时也支持自定义序列化方法序列

9、化管理额外存储和管理数据编解码信息，包括各种对序列化数据的压缩或加密策略编码和加密支持基于 Pravega Key-Value Table 功能进行存储底层实现不仅限用于 Pravega 项目，可以被用于通用的存储系统的 schema 管理解决方案通用性项目介绍项目介绍系统架构系统架构Schema RegistryAdminWritersReadersGRPCRESTGroupSchema TypeCompatibility StrategyVersioned HistoryGroup:Scope/Stream1 AvroFullv1,v2,v3,vKGroup:Scope/Stream2Pr

10、otobufDenyAllv1Group:Scope/Stream3JsonAllowAnyv1,v2,v3,vNPravegaSegment Store1 byteProtocol version4 bytesEncoding IDEncoded bytes#4Catalog 支持支持Flink CatalogFLIP-30：基于 Flink 与 Hive Metastore 集成，使得 Hive 中的元数据能与 Flink Table 抽象互通将 Hive 元数据信息抽象化，例如数据库、表、分区、视图以及数据库，额外的函数、分区信息等。提供相应的接口用以对接通用的外部 connector

11、系统Catalog 与与Pravega 的映射关系的映射关系FlinkPravegaPravega Schema RegistryDatabaseScopeNamespaceTableStream(With Schema)GroupView/Partition/Function/Statistic/TableStream 中的流数据SchemaSchema RegistryJson+type:object,properties:lon:type:number,rideTime:type:string5 bytes+“lon”:7,“rideTime”:“2020-12-30 12:13:14.

12、123”lonrideTime72020-12-30 12:13:14.123Catalog 集成集成实现难点及解决方案:难点难点1：Schema 处理处理FLINK-19098 RowData 与 Json/Avro 数据转换变为 public 类数据序列化和 Schema 的转化引入 serialization.format 选项，可进一步定制应用于整个 catalog 的序列化方法，如：json.timestamp-format.standard开发新的 FormatFactory难点难点2：同时支持：同时支持Avro 和和Json 格式格式难点难点3：序列化方式与标准不同：序列化方式与

13、标准不同“pravega-registry”Format FactoryEvent 开头的 magic bytes 导致了无法使用 Flink 现成的 Json/Avro format使用 Schema registry 根据存储信息提供的序列化器将 Catalog Table 中与序列化相关的参数沿袭到 format factory 参数byteJsonNode/GenericRecordRowDataSchema Registry APIFlink API未来规划未来规划Protobuf 格式的支持Upsert table source 的支持自定义序列化的支持FLINK-18202 Introduce Protobuf format#5DemoDemo1