溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
這篇文章主要介紹“DataStreamReader和DataStreamWriter怎么使用”，在日常操作中，相信很多人在DataStreamReader和DataStreamWriter怎么使用問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”DataStreamReader和DataStreamWriter怎么使用”的疑惑有所幫助！接下來，請跟著小編一起來學習吧！
流的讀取是從DataStreamReader和DataStreamWriter開始的。
DataStreamReader是生成流讀取者的入口所在，關鍵方法是load。這段代碼很關鍵，所以把全部代碼先貼出來，慢慢分析。
def load(): DataFrame = {
val ds = DataSource.lookupDataSource(source, sparkSession.sqlContext.conf).
getConstructor().newInstance()
val v1DataSource = DataSource(
sparkSession,
userSpecifiedSchema = userSpecifiedSchema,
className = source,
options = extraOptions.toMap)
val v1Relation = ds match {
case _: StreamSourceProvider => Some(StreamingRelation(v1DataSource))
case _ => None
}
ds match {
case provider: TableProvider =>
val sessionOptions = DataSourceV2Utils.extractSessionConfigs(
source = provider, conf = sparkSession.sessionState.conf)
val options = sessionOptions ++ extraOptions
val dsOptions = new CaseInsensitiveStringMap(options.asJava)
val table = userSpecifiedSchema match {
case Some(schema) => provider.getTable(dsOptions, schema)
case _ => provider.getTable(dsOptions)
}
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Implicits._
table match {
case _: SupportsRead if table.supportsAny(MICRO_BATCH_READ, CONTINUOUS_READ) =>
Dataset.ofRows(
sparkSession,
StreamingRelationV2(
provider, source, table, dsOptions, table.schema.toAttributes, v1Relation)(
sparkSession))
// fallback to v1
// TODO (SPARK-27483): we should move this fallback logic to an analyzer rule.
case _ => Dataset.ofRows(sparkSession, StreamingRelation(v1DataSource))
}
case _ =>
// Code path for data source v1.
Dataset.ofRows(sparkSession, StreamingRelation(v1DataSource))
}
}有好多分支，重要的是區分開V1和V2。
V1用的邏輯關系是StreamingRelation；而V2用的邏輯關系是StreamingRelationV2。這里先看看他們對應的物理計劃是什么？
在SparkStrategies.scala文件中，定義了物理計劃：
/**
* This strategy is just for explaining `Dataset/DataFrame` created by `spark.readStream`.
* It won't affect the execution, because `StreamingRelation` will be replaced with
* `StreamingExecutionRelation` in `StreamingQueryManager` and `StreamingExecutionRelation` will
* be replaced with the real relation using the `Source` in `StreamExecution`.
*/
object StreamingRelationStrategy extends Strategy {
def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
case s: StreamingRelation =>
StreamingRelationExec(s.sourceName, s.output) :: Nil
case s: StreamingExecutionRelation =>
StreamingRelationExec(s.toString, s.output) :: Nil
case s: StreamingRelationV2 =>
StreamingRelationExec(s.sourceName, s.output) :: Nil
case _ => Nil
}
}物理計劃都是StreamingRelationExec，StreamingRelationExec的代碼其實啥都沒實現，所以最后其實看代碼注釋StreamingRelationExec也不是真正的物理計劃。
這里先記得相關的類ContinuousExecution和MicroBatchExecution。一時找不到怎么執行到具體的物理計劃ContinuousExecution和MicroBatchExecution的，我們就試試反推把。先看看ContinuousExecution的代碼。
StreamExecution是抽象類。其抽象方法runActivatedStream是執行具體的連續流讀取任務的，子類會重寫該函數。
runStream方法封裝了runActivatedStream方法，額外加了些事件通知等處理機制，知道這一點就行了。
這里先嘗試看看StreamingQueryManager是干什么用的，看注釋應該是管理所有的StreamingQuery的。
private def createQuery(...): StreamingQueryWrapper ={
(sink, trigger) match {
case (table: SupportsWrite, trigger: ContinuousTrigger) =>
new StreamingQueryWrapper(new ContinuousExecution(
sparkSession,
userSpecifiedName.orNull,
checkpointLocation,
analyzedPlan,
table,
trigger,
triggerClock,
outputMode,
extraOptions,
deleteCheckpointOnStop))
case _ =>
if (operationCheckEnabled) {
UnsupportedOperationChecker.checkForStreaming(analyzedPlan, outputMode)
}
new StreamingQueryWrapper(new MicroBatchExecution(
sparkSession,
userSpecifiedName.orNull,
checkpointLocation,
analyzedPlan,
sink,
trigger,
triggerClock,
outputMode,
extraOptions,
deleteCheckpointOnStop))
}
}對于連續流，返回一個：
new StreamingQueryWrapper(new ContinuousExecution））
StreamingQueryWrapper的作用，就是將StreamingQuery封裝成可序列化的，別的和StreamingQuery沒什么區別。這里對于連續流就是包裝了ContinuousExecution。
ContinuousExecution看名稱應該是對應連續流的物理執行計劃的，繼承自StreamExecution（抽象類）。看看主要代碼其實就是重寫了runActivatedStream方法。
override protected def runActivatedStream(sparkSessionForStream: SparkSession): Unit = {
val stateUpdate = new UnaryOperator[State] {
override def apply(s: State) = s match {
// If we ended the query to reconfigure, reset the state to active.
case RECONFIGURING => ACTIVE
case _ => s
}
}
do {
runContinuous(sparkSessionForStream)
} while (state.updateAndGet(stateUpdate) == ACTIVE)
stopSources()
}真正的執行邏輯代碼在私有方法runContinuous中，這里就不詳細展開了，知道了主要流程就可以了。
下面就是要看看ContinuousExecution到底是在哪里被從邏輯計劃轉換到物理計劃的。
搜索全文，找到了StreamingQueryManager.scala這個文件。對了，就是從上面的StreamingQueryManager找到這個ContinuousExecution。
DataStreamWriter是真正觸發流計算開始啟動執行的地方。
start()方法得到要給StreamingQuery，方法里的關鍵代碼片段：
df.sparkSession.sessionState.streamingQueryManager.startQuery(
extraOptions.get("queryName"),
extraOptions.get("checkpointLocation"),
df,
extraOptions.toMap,
sink,
outputMode,
useTempCheckpointLocation = source == "console" || source == "noop",
recoverFromCheckpointLocation = true,
trigger = trigger)跟蹤進去到了StreamingQueryManager，看它的startQuery方法。
startQuery方法分為幾步：
調用createQuery方法返回StreamingQuery。
val query = createQuery( userSpecifiedName, userSpecifiedCheckpointLocation, df, extraOptions, sink, outputMode, useTempCheckpointLocation, recoverFromCheckpointLocation, trigger, triggerClock)
query就是StreamingQueryWrapper，就是類似這樣的代碼：
new StreamingQueryWrapper(new ContinuousExecution））
2、啟動上一步的query
try {
query.streamingQuery.start()
} catch {
}這里的代碼直接調用到StreamingQuery的父類StreamExecution的start方法。代碼定義：
def start(): Unit = {
logInfo(s"Starting $prettyIdString. Use $resolvedCheckpointRoot to store the query checkpoint.")
queryExecutionThread.setDaemon(true)
queryExecutionThread.start()
startLatch.await() // Wait until thread started and QueryStart event has been posted
}queryExecutionThread線程的定義又是這樣的：
val queryExecutionThread: QueryExecutionThread =
new QueryExecutionThread(s"stream execution thread for $prettyIdString") {
override def run(): Unit = {
sparkSession.sparkContext.setCallSite(callSite)
runStream()
}
}最后在線程中啟動runStream這個私有方法。
3、返回query
最后返回query，注意這里的query在上面的代碼中已經start運行了。
到此，關于“DataStreamReader和DataStreamWriter怎么使用”的學習就結束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續學習更多相關知識，請繼續關注億速云網站，小編會繼續努力為大家帶來更多實用的文章！
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
