結構化Kafka sql的代碼框架是怎樣的

本篇文章給大家分享的是有關結構化Kafka sql的代碼框架是怎樣的，小編覺得挺實用的，因此分享給大家學習，希望大家閱讀完這篇文章后可以有所收獲，話不多說，跟著小編一起來看看吧。

結構化流的典型應用是持續的讀取kafka流。實現機制從SparkSession的readStream開始，readStream就是DataStreamReader：

def readStream: DataStreamReader = new DataStreamReader(self)

下面從DataStreamReader開始。可以想象得到，最終肯定是生成一個RDD來持續讀取kafka流數據。

例子：

// Create DataFrame representing the stream of input lines from connection to localhost:9999
val lines = spark.readStream
  .format("socket")
  .option("host", "localhost")
  .option("port", 9999)
  .load()

分兩步：找到TableProvider；找到SupportRead然后生成StreamingRelationV2。

最后用StreamingRelationV2來調用Dataset.ofRows返回DataFrame，DataFrame就是Dataset[Row]。

下面首先要看看TableProvider接口和SupportRead接口是啥東東。

TableProvider

TableProvider接口未找到在哪里定義。

KafkaSourceRDD

先看看kafkaSourceRDD這個類，這是基礎類，最基礎的來讀取kafka數據的RDD，入參包含一個offsetRange，表示讀取kafka數據的區間范圍。如果是Kafka.lastest則可以表示永久讀取kafka。

既然是RDD，那么最重要的方法就是compute方法了，代碼不解析了很簡單，就是用Kafka的API來讀取kafka分區的數據，形成RDD。

KafkaSource

KafkaSource顧名思義就是Kafka的讀取者。

KafkaSource的父類是Source，最重要的方法是：getOffset和getBatch。

getBatch返回DataFrame，那么getBatch又是怎么返回DataFrame的呢？看代碼就知道原來是通過創建KafkaSourceRDD來達到生成DataFrame的目的的。所以可以認為KafkaSource是KafkaSourceRDD的一種封裝形式罷了。

KafkaSourceProvider

The provider class for all Kafka readers and writers。這個類是用來生成各種各樣的Kafka的讀取者和寫入者的，比較重要，先看看這個類的定義：

private[kafka010] class KafkaSourceProvider extends DataSourceRegister

    with StreamSourceProvider

    with StreamSinkProvider

    with RelationProvider

    with CreatableRelationProvider

    with TableProvider

    with Logging 

繼承了很多的特性或接口。比如：StreamSourceProvider、TableProvider、RelationProvider等等。我們這里就看看和讀相關的特性吧，和寫相關的不看了（道理差不多）。

（1）createSource

createSource方法返回Source，看代碼其實返回的是KafkaSource，KafkaSource前面已經說過了，這里就不涉及了。

（2）createRelation

createRelation返回BaseRelation，實際返回的是KafkaRelation。

KafkaRelation繼承BaseRelation，重寫父 類的buildScan方法，buildScan方法返回KafkaSourceRDD作為RDD[Row]。

（3）KafkaTable

KafkaTable繼承Table并且繼承SupportsRead特性，其定義：

class KafkaTable(includeHeaders: Boolean) extends Table with SupportsRead with SupportsWrite 

里面輾轉反側看看如何生成ContinuousStream，主要是方法toContinuousStream，返回的ContinuousStream就是KafkaContinuousStream。

（4）KafkaContinuousStream

KafkaContinuousStream繼承自ContinuousStream，具體的看代碼，最后反正都是調用了Kafka的API來讀取數據，所不同的只是外部表現形式的不同罷了。

以上就是結構化Kafka sql的代碼框架是怎樣的，小編相信有部分知識點可能是我們日常工作會見到或用到的。希望你能通過這篇文章學到更多知識。更多詳情敬請關注億速云行業資訊頻道。