Flink SQL实战演练之自定义Table Format

Posted 2023-04-06

参考技术A 简介：接着上次Flink CDC继续聊，提到这块，不得不重点说下canal-json format了，canal json format对json format进行了封装，负责把binlog形式的json数据转化成了Flink能够识别的RowData数据，当然也包括数据的描述信息封装类RowType。笔者想实现根据RowKind进行数据的过滤，目前可以通过修改canal format的源数据来实现，也可以通过将changelog流以changelog json的形式回写Kafka。

基于目前对table format的了解，这里自定义event json format，用来处理事件流数据，因为事件流字段不固定，可能只有少部分字段是固定的，其他字段都是扩展的，所以笔者想实现用户自定义schema指定公共字段，然后其他字段以json的行为存在metadata中的default字段中。

Table Format作为Connector组件单独用于序列化和反序列化内部数据的模块而单独存在，多个Connector可以公用。自定义Table Format可以让大家更好的理解Flink SQL时如何将外部数据转化为内部可以识别的RowData数据结构的，从而在排查问题的时候能准确定位到具体位置。

Flink Table API & SQL 自定义 Scalar 标量函数

1. 什么是标量函数

Scalar Function 也被称为标量函数，将 0 个、1个或多个标量值映射为一个新的标量值。输入与输出是一对一的关系，即读入一行数据，写出一条输出值。在自定义标量函数时，用户需要确认 Flink 内部是否已经实现相应的标量函数，如果已经实现则可以直接使用；如果没有实现，那么在注册自定义函数过程中，需要和内置的其他标量名称区分开，否则会导致注册函数失败，影响应用的正常执行。Flink 常见的内置标量函数有 DATE()、UPPER()、LTRIM() 等。

2. 定义标量函数

定义 Scalar Function 需要继承 org.apache.flink.table.functions.ScalarFunction 类。实现函数的类必须声明为 public、不能是抽象类，并且可以全局访问。因此，不允许使用非静态内部类或者匿名类。如果要在 Catalog 中存储用户自定义的函数，那么该类必须具有一个默认构造函数并且必须在运行时可实例化：

public class AddScalarFunction extends ScalarFunction 
   
  .