一、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 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
二、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