建站
咨詢
售后
建站咨詢
新聞中心
本篇文章給大家分享的是有關結構化Kafka sql的代碼框架是怎樣的，小編覺得挺實用的，因此分享給大家學習，希望大家閱讀完這篇文章后可以有所收獲，話不多說，跟著小編一起來看看吧。
創(chuàng)新互聯(lián)專注于企業(yè)成都營銷網站建設、網站重做改版、湖南網站定制設計、自適應品牌網站建設、H5技術、商城建設、集團公司官網建設、成都外貿網站建設公司、高端網站制作、響應式網頁設計等建站業(yè)務，價格優(yōu)惠性價比高，為湖南等各大城市提供網站開發(fā)制作服務。
結構化流的典型應用是持續(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來調用Dataset.ofRows返回DataFrame，DataFrame就是Dataset[Row]。
下面首先要看看TableProvider接口和SupportRead接口是啥東東。
TableProvider
TableProvider接口未找到在哪里定義。
KafkaSourceRDD
先看看kafkaSourceRDD這個類，這是基礎類，最基礎的來讀取kafka數(shù)據(jù)的RDD，入參包含一個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等等。我們這里就看看和讀相關的特性吧，和寫相關的不看了（道理差不多）。
（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來讀取數(shù)據(jù)，所不同的只是外部表現(xiàn)形式的不同罷了。
以上就是結構化Kafka sql的代碼框架是怎樣的，小編相信有部分知識點可能是我們日常工作會見到或用到的。希望你能通過這篇文章學到更多知識。更多詳情敬請關注創(chuàng)新互聯(lián)行業(yè)資訊頻道。
新聞標題：結構化Kafkasql的代碼框架是怎樣的
鏈接URL：http://www.dlmjj.cn/article/jcscjs.html
