Flink-源码Bug修复和二次开发实践

一时失言乱红尘 2022-12-28 01:37 144阅读 0赞

# Flink-源码Bug修复和二次开发实践 #

# 1 Bug修复 #

*  解决Flink Sql DDL 中使用`time`作为字段关键字某些场景报错的Bug  
    flink 1.10
 *  解决Json反序列化时，如果内容带`\`时，`TextNode`直接转`ObjectNode`接续报错的Bug

# 2 新增功能 #

## 2.1 KuduConnector ##

参考了bahir-flink ，基于Flink 1.11 ，围绕最新的RowData类型，实现以下功能：

*  KuduDynamicSink  
    可直接Flink Sql流式写入Kudu。还有继续优化空间。
 *  KuduDynamicSource  
    实现了
    
     *  `ScanTableSource`,
     *  `SupportsProjectionPushDown`
     *  `LookupTableSource`,
        
         *  支持了实现自`AsyncTableFunction`的`AsyncKuduRowDataLookupFunctionV2`，基于Flink 异步IO，用于异步读取Kudu表数据。  
            还加入了两个缓存：
            
             *  `Cache<RowData, List<RowData>>`用来缓存join keys对应的数据；
             *  `Cache<RowData, CompletableFuture<List<RowData>>>`用于缓存join keys对应的`CompletableFuture`，最大限度减少短时间内同key数据对kudu的访问
             *  以上Cache基于Guava Cache，后期准备对比测试Caffeine Cache性能以决定是否替换
         *  目前可达到几万条/秒，同时又不对kudu造成较大压力

## 2.2 ElasticsearchConnector ##

基于Flink 1.11，围绕最新的RowData类型，实现以下功能：

*  Elasticsearch6DynamicSource
    
     *  实现了 `LookupTableSource`, `SupportsProjectionPushDown`  
        主要用于Elasticsearch维表join
        
         *  支持了实现自`AsyncTableFunction`的`AsyncKuduRowDataLookupFunctionV2`，基于Flink 异步IO，用于异步读取Kudu表数据。  
            还加入了两个缓存：
            
             *  `Cache<RowData, List<RowData>>`用来缓存join keys对应的数据；
             *  `Cache<RowData, CompletableFuture<List<RowData>>>`用于缓存join keys对应的`CompletableFuture`，最大限度减少短时间内同key数据对kudu的访问

## 2.3 Parquet+LZO ##

Flink 1.10中，加入了对Parquet格式下LZO压缩的支持，且已经投入生产，节约过半硬盘空间。

## 2.4 Kafka CDC MaxWell Format ##

可实现MaxWell解析的Binlog推送到Kafka后的CDC对接，转为Flink内部统一识别的`RowData`。可支持增删改查。

# 附录 #

## 打包命令 ##

clean package -T 2C -Dfast -Dmaven.compile.fork=true 
    -DskipTests -Dscala-2.11 -Dmaven.javadoc.skip=true -Dcheckstyle.skip=true -e -X -f pom.xml