Flink table 转datastream SINK

「爱情、让人受尽委屈。」 2022-09-15 05:59 133阅读 0赞

# 一、table source #

## 1、TableSource源码 ##

trait TableSource\[T\] \{ //TableSource 是一个接口总共有三个方法

def getReturnType: TypeInformation\[T\] //获取返回的DataStream或者 DataSet的类型

def getTableSchema: TableSchema //获取table表结构。table里面有哪些字段，字段是什么类型

def explainSource():String =TableConnectorUtil.generateRuntimeName(getClass,getTableSchema.getFieldNames) //table source 的一个描述，不重要

## 2、TableSource实现 ##

trait BatchTableSource\[T\] extends **TableSource\[T\]** //离线table

trait **StreamTableSource**\[T\] extends **TableSource\[T**\] //实时table

BatchTableSource和StreamTableSource 就是在TableSource的基础上扩展了一个获取批数据和流数据的方法，如下

批：def getDataSet(execEnv: ExecutionEnvironment): DataSet\[T\]

流：def getDataStream(execEnv: StreamExecutionEnvironment): DataStream\[T\]

批不做重点解释。基于StreamTableSource的实现这里只介绍kafka相关,后续如果需要扩展其他tableSoure 比如hbase 肯定也需要实现 StreamTableSource，后续文集也会给出对应的代码实现

public abstract class **KafkaTableSourceBase** implements **StreamTableSource**,... \{\}

public class Kafka09TableSource extends **KafkaTableSourceBase**\{\} //kafka0.9 的tableSource实现

public class Kafka010TableSource extends **KafkaTableSourceBase\{\}**  //kafka0.10 的tableSource实现

可以看下Kafka010TableSource 有如下方法

protected **FlinkKafkaConsumerBase** createKafkaConsumer(...) \{

return new FlinkKafkaConsumer010<>(topic, deserializationSchema, properties);

看看**FlinkKafkaConsumerBase**是什么玩意

public abstract class **FlinkKafkaConsumerBase** extends **RichParallelSourceFunction**

这样可以看到其实 tableSource 内部也是会初始化**SourceFunction**

**[SourceFunction 详解见 https://www.jianshu.com/p/d6427dcf7ea2][SourceFunction _ https_www.jianshu.com_p_d6427dcf7ea2]**

# 二、table sink #

## 1、TableSink源码 ##

trait TableSink\[T\] \{

def getOutputType: TypeInformation\[T\] //获取输出的数据类型

def getFieldNames: Array\[String\] //获取输出的字段

def getFieldTypes: Array\[TypeInformation\[\_\]\] //获取输出的字段类型

def configure(fieldNames: Array\[String\], fieldTypes: Array\[TypeInformation\[\_\]\]): TableSink\[T\] //根据当前的TableSink复制一个TableSink

## 2、TableSink实现 ##

trait BatchTableSink\[T\] extends **TableSink**\[T\] //离线table

trait **StreamTableSink**\[T\] extends **TableSink**\[T\] //实时table

BatchTableSource和StreamTableSource 就是在TableSource的基础上扩展了一个将批数据和流数据的方法发送到sink组件的方法，具体是什么sink需要自己去实现。如下

批：def emitDataSet(dataSet: DataSet\[T\]):Unit

流：def emitDataStream(dataStream: DataStream\[T\]):Unit

批不做重点解释。StreamTableSource 又有三种实现如下

trait AppendStreamTableSink\[T\] extends StreamTableSink\[T\]

trait RetractStreamTableSink\[T\] extends StreamTableSink\[JTuple2\[JBool, T\]\]

trait UpsertStreamTableSink\[T\] extends StreamTableSink\[JTuple2\[JBool, T\]\]

对于只增加记录的场景，可以使用AppendStreamTableSink。

对于有修改的场景，删除的场景那必须使用 RetractStreamTableSink，或者 UpsertStreamTableSink。 这2个接口发送到sink组件的是 JTuple2\[JBool, T\] 这个类型。元组的第一个元素用来表示这条记录是删除还是新增（false 代表删除，true代表新增）。

那什么时候会有修改删除呢？比如是对group by语句进行Sink. 比如单词统计，统计每个词出现的次数。 select count(\*),word from workcount group by word; 由于数据不像批处理那样一次拉取到所有数据进行一次 group by产生一个准确的结果。流处理中数据是源源不断的进来，也就意味着group by 会持续进行。先来一个单词 ‘A’,那么'A'的次数为 'A':1 ；后面又来一个单词 ‘A’;又会进行group by 那么'A'的次数为 'A':2;这样就会存在2个结果。显然第一次统计的结果 'A':1应该要被删除。所以采用RetractStreamTableSink 对于第2次group by的时候回返回2条数据。（false,（A,1））和（true,（A,2））标识第一条要删掉也叫撤回。第二条新增。

table 转换成 dataStream 时有2种Api toRetractStream 和 toAppendStream例如

val result:Table = tableEnv.sqlQuery("select \* from json ")

1、tableEnv.toRetractStream\[Row\](result).print()

t2、ableEnv.toAppendStream\[Row\](result).print()

对于有group by的语句就不能toAppendStream 来进行转换。

在kafka中默认实现是AppendStreamTableSink

public abstract class **KafkaTableSinkBase** implements **AppendStreamTableSink** //kafka tableSink 的实现基类

public class Kafka09TableSink extends **KafkaTableSinkBase** //kafka0.9 的tableSinkl实现

public class Kafka010TableSink extends **KafkaTableSinkBase** //kafka0.10 的tableSinkl实现

我觉得kafka实现是RetractStreamTableSink 还是挺有必要的。虽然kafka自身是不支持删除和修改的，但是我可以把数据发送给下游，让下游根据ture，false 标识来做决策进行删除新增。后面文章会给出对应的代码实现。后续如果需要新增其他组件的tableSink 比如 hbase 就需要实现AppendStreamTableSink、RetractStreamTableSink、UpsertStreamTableSink 其中的一个或多个。后续文集也会给出对应的代码实现。

可以看一下Kafka010TableSink 有如下方法

protected **FlinkKafkaProducerBase** createKafkaProducer(...)\{

return new FlinkKafkaProducer010<>(...)

**FlinkKafkaProducerBase** 是啥玩意呢？

public abstract class **FlinkKafkaProducerBase** extends **RichSinkFunction** implements CheckpointedFunction\{\}

这样可以看到其实 tableSink 内部也是会初始化**SinkFunction.**

# **三、结论** #

通过上文我们得到几个重要的信息，这里在重申单独列出一下。

1、在流中要自定义**TableSoure** 就必须实现**StreamTableSource**

2、**StreamTableSource**的实现中必须有入口初始化**SourceFunction。**也就是说在初始化**StreamTableSource**的时候最终会初始化**SourceFunction**

3、在自定义**TableSoure** 时除了要自实现**StreamTableSource**也要实现对应的**SourceFunction**

4、在流中要自定义**TableSink** 就必须实现**AppendStreamTableSink、RetractStreamTableSink、UpsertStreamTableSink** 其中的一个或多个

5、**AppendStreamTableSink、RetractStreamTableSink、UpsertStreamTableSink**的实现中必须有入口初始化**SinkFunction。**也就是说在初始化**AppendStreamTableSink、RetractStreamTableSink、UpsertStreamTableSink**的时候最终会初始化**SinkFunction**

6、在自定义**TableSink**时除了要自实现**AppendStreamTableSink、RetractStreamTableSink、UpsertStreamTableSink**也要实现对应的**SinkFunction**

7、怎么自定义实现**SourceFunction，SinkFunction** 两者是相似的，可以参考下文连接

**[SourceFunction 详解见 https://www.jianshu.com/p/d6427dcf7ea2][SourceFunction _ https_www.jianshu.com_p_d6427dcf7ea2]**

[SourceFunction _ https_www.jianshu.com_p_d6427dcf7ea2]: https://www.jianshu.com/p/d6427dcf7ea2