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

創(chuàng)新互聯(lián)專注于企業(yè)成都營銷網(wǎng)站建設(shè)、網(wǎng)站重做改版、湖南網(wǎng)站定制設(shè)計、自適應(yīng)品牌網(wǎng)站建設(shè)、H5技術(shù)、商城建設(shè)、集團公司官網(wǎng)建設(shè)、成都外貿(mào)網(wǎng)站建設(shè)公司、高端網(wǎng)站制作、響應(yīng)式網(wǎng)頁設(shè)計等建站業(yè)務(wù)，價格優(yōu)惠性價比高，為湖南等各大城市提供網(wǎng)站開發(fā)制作服務(wù)。

結(jié)構(gòu)化流的典型應(yīng)用是持續(xù)的讀取kafka流。實現(xiàn)機制從SparkSession的readStream開始，readStream就是DataStreamReader：

def readStream: DataStreamReader = new DataStreamReader(self)

下面從DataStreamReader開始?？梢韵胂蟮玫?，最終肯定是生成一個RDD來持續(xù)讀取kafka流數(shù)據(jù)。

例子：

// 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來調(diào)用Dataset.ofRows返回DataFrame，DataFrame就是Dataset[Row]。

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

TableProvider

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

KafkaSourceRDD

先看看kafkaSourceRDD這個類，這是基礎(chǔ)類，最基礎(chǔ)的來讀取kafka數(shù)據(jù)的RDD，入?yún)粋€offsetRange，表示讀取kafka數(shù)據(jù)的區(qū)間范圍。如果是Kafka.lastest則可以表示永久讀取kafka。

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

KafkaSource

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

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

getBatch返回DataFrame，那么getBatch又是怎么返回DataFrame的呢？看代碼就知道原來是通過創(chuàng)建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等等。我們這里就看看和讀相關(guān)的特性吧，和寫相關(guān)的不看了（道理差不多）。

（1）createSource

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

（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 

里面輾轉(zhuǎn)反側(cè)看看如何生成ContinuousStream，主要是方法toContinuousStream，返回的ContinuousStream就是KafkaContinuousStream。

（4）KafkaContinuousStream

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

以上就是結(jié)構(gòu)化Kafka sql的代碼框架是怎樣的，小編相信有部分知識點可能是我們?nèi)粘９ぷ鲿姷交蛴玫降?。希望你能通過這篇文章學(xué)到更多知識。更多詳情敬請關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道。