续  一个基于flinkSql 的实时计算平台 https://www.jianshu.com/p/db1a89e6fa85 文中 剖析fink中kafkaTbaleSource的实现

此文想看看Kafka 在 flink table 中的实现。

首先再flink1.7中TbaleSource定义的流程是，先定义 tableFactory 再定义 tableSource 再定义 sourceFunction. 我们就按照这个套路来看看源代码

一、Kafka010TableSourceSinkFactory

Kafka010TableSourceSinkFactory 

看一个方法createKafkaTableSource（）。这个方法会再初始化查找到tableFactory 之后就会被调用。

至于查找tableFactory  的过程是怎样的可以参考 flink table factory基础知识 https://www.jianshu.com/p/6b755ed1a5bb 一文去跟一下源代码。对createKafkaTableSource的参数做下解释

TableSchema schema ：kafka的数据最终都会被映射到table中的一行记录。这个schema 就是用来描述table结构用的，标识kafka中数据的列名和数据类型

Optional<String> proctimeAttribute：时间属性来标识是用时间时间还是处理时间，默认是处理时间

List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors：时间描述。如果有类似窗口的作业，基本上都会有指定时间，这个里就是描述你指定的时间是那个字段，叫什么名字，水位的定义等

String topic, kafka topic名字

Properties properties, 链接kafka的一些属性比如zk地址等

DeserializationSchema deserializationSchema, 这个是反序列化器，反序列化的查找过程同查找tableFactory 的过程。

StartupMode startupMode： 就是消费kafka offset的模式。有EARLIEST，LATEST，默认是从上一次消费位置开始消费

Map<KafkaTopicPartition, Long> specificStartupOffsets: kafka offset的模式可以选择指定从某个offset位置开始消费，如果是这种模式 specificStartupOffsets 就是用来标识每个topic partion 要开始消费的offset位置。

createKafkaTableSource（） 啥事也么干。开门见山的去 new 了一下 Kafka010TableSource()。参数原封不动的传给Kafka010TableSource。 下面来看下 Kafka010TableSource

二、Kafka010TableSource

Kafka010TableSource

这个类也很明了。三个方法，2个构造方法，一个创建kafkaCounsumer的方法。构造方法可以忽略无非就是再初始化对象的时候对对象的成员变量进行初始化,

看到createKafkaConsumer 这个方法的名字可以感觉到这是去创建一个kafka的消费者。本文是看kafkaTbaleSource的实现。数据来源是kafka，flink是作为一个消费端，接受到数据然后作为数据处理的soure.先不看里面的Consumer代码长啥样，先看下他在哪里被调用。

createKafkaConsumer()来自于父类，父类此方法是抽象的，因为对于不同版本的kafka有不一样的实现。createKafkaConsumer()这个方法在父类的 getKafkaConsumer 方法中被调用如下

protected FlinkKafkaConsumerBase   getKafkaConsumer(...) {

    kafkaConsumer =createKafkaConsumer(...)

}

getKafkaConsumer  被getDataStream 方法调用如下

public DataStream getDataStream(StreamExecutionEnvironment env) {

    DeserializationSchema deserializationSchema = getDeserializationSchema();

   FlinkKafkaConsumerBase kafkaConsumer = getKafkaConsumer(topic, properties, deserializationSchema);

  return env.addSource(kafkaConsumer).name(explainSource());

}

 env.addSource(kafkaConsumer) 看到这句应该熟悉flink api的同学会很熟悉。这不是就是添加一个soure的做法吗？ 我们不用flink sql的时候也就是这么干的. 这说明sql的内部也是用stream api去实现的。只是在之上做了一些包装把数据转化成表。把逻辑封装成sql语句。

getDataStream 方法来自于StreamTableSource 如下

trait StreamTableSource[T] extends TableSource[T] {

  def getDataStream(execEnv: StreamExecutionEnvironment): DataStream[T]

}

这是一个获取流的方法。在flink在将table api转化成执行计划的过程会被调用。

再来捋一下上面的流程。在调用 ConnectTableDescriptor.registerTableSource(tableName)的时候会去查找tableFactory。查找到kafka相关的factory 比如 Kafka010TableSourceSinkFactory。之后会调用tableFactory的createStreamTableSource方法。createStreamTableSource 这个方法会调用 createKafkaTableSource。进行初始化Kafka010TableSource。 之后再解析执行计划的时候会调用Kafka010TableSource 的 getDataStream 方法。这个方法会调用 getKafkaConsumer 方法启动一个kafka的消费者，作为dataStream。最后通过env.addSource(kafkaConsumer)方法为flink添加soure。 这就是整个Kafka Table 的执行过程。

三、FlinkKafkaConsumerBase

前面说到了获取kafkaConsumer，那他到底是个啥玩意呢？先看下下基类FlinkKafkaConsumerBase

FlinkKafkaConsumerBase

看到FlinkKafkaConsumerBase 是继承  RichParallelSourceFunction 应该要知道这是给flink添加一个streamSource的基本方式就是要继承 RichParallelSourceFunction 、 RichSourceFunction 、ParallelSourceFunction、SourceFunction 。 这块不明白的请参考flink soruce详解 https://www.jianshu.com/p/d6427dcf7ea2

明白RichParallelSourceFunction是个啥玩意之后。只需要着重看里面的run 方法。再run方法中做了很多初始化工作。比如createFetcher这就是创建真正的消费者。最后会调用fetcher 的runFetchLoop 方法。 里面就是一个死循环 去消费kafka数据最后把数据发送给flink.如下截图。

flink kafka 消费者