資訊專欄INFORMATION COLUMN
點(diǎn)擊上方“IT那活兒”公眾號(hào)，關(guān)注后了解更多內(nèi)容，不管IT什么活兒，干就完了?。。?/strong>
應(yīng)用程序中只有調(diào)用了start()才能真正開(kāi)始執(zhí)行查詢，然后返回StreamingQuery對(duì)象，您可以使用該對(duì)象來(lái)管理查詢。先舉一些例子：
foreach和foreachBatch操作允許您對(duì)流式查詢的輸出應(yīng)用任意操作和寫入邏輯。它們的用例稍有不同——foreach允許在每一行上自定義寫入邏輯，而foreachBatch允許在每個(gè)微批的輸出上執(zhí)行任意操作和自定義邏輯。讓我們更詳細(xì)地了解它們的用法。
2.1 foreachBatch
foreachBach允許你對(duì)流式查詢的每個(gè)微批的輸出數(shù)據(jù)指定執(zhí)行的函數(shù)，自spark2.4，scala，java，python都支持這一點(diǎn)，他有兩個(gè)參數(shù)，微批數(shù)據(jù)集和微批的唯一ID。
使用foreachBatch可以做以下工作：
重用批處理數(shù)據(jù)源-對(duì)于許多存儲(chǔ)系統(tǒng)，可能還沒(méi)有可用的流式接收器，但可能已經(jīng)存在批處理查詢的接收器，所以使用foreachBatch可以再每個(gè)微批次輸出中使用批接收器。
寫入多個(gè)存儲(chǔ)器：可能你需要將流式查詢的輸出寫入到多個(gè)位置，則只需多次寫入即可，但是這樣可能導(dǎo)致重新計(jì)算（包括可能重新讀取數(shù)據(jù)）。為了避免重新計(jì)算，應(yīng)該緩存數(shù)據(jù)集，將其寫入多個(gè)存儲(chǔ)器后再取消緩存，例如：
批處理中的許多操作在流式查詢中不支持，因?yàn)閟park不支持在這些情況下生成增量計(jì)劃，使用foreachBatch可以將流式查詢轉(zhuǎn)換成了一個(gè)個(gè)微批來(lái)處理，但是你必須自己考慮執(zhí)行這些操作端到端語(yǔ)義。
注意：
默認(rèn)情況下foreachBatch只提供至少一次寫入保證，但是你可以使用batchID作為消除重復(fù)的方法，并獲得一次寫入保證。
foreachBatch不適用于連續(xù)處理模式，因?yàn)樗旧弦蕾囉谖⑴幚?，如果以連續(xù)模式寫入數(shù)據(jù)，可以使用foreach。
2.2 foreach
如果不存在相應(yīng)的批處理數(shù)據(jù)接收器，或者不存在連續(xù)處理模式，則可以使用foreach來(lái)自定義編寫器邏輯，你可以將數(shù)據(jù)寫入邏輯分為三個(gè)方法，open，process和close。
自spark2.4 scala，java，python都支持這一點(diǎn)。
流查詢啟動(dòng)后，spark按以下方式調(diào)用函數(shù)或?qū)ο蟮姆椒ǎ?/span>
此對(duì)象的一個(gè)副本負(fù)責(zé)查詢中單個(gè)任務(wù)生成的所有數(shù)據(jù)，也就是說(shuō)，一個(gè)實(shí)例負(fù)責(zé)處理以分布式方式生成的數(shù)據(jù)的一個(gè)分區(qū)。
此對(duì)象必須是可序列化的，因?yàn)槊總€(gè)任務(wù)將獲取此對(duì)象的副本，需要進(jìn)行反序列化，強(qiáng)烈建議任何初始化一定在調(diào)用open()方法之后完成，意味著已經(jīng)準(zhǔn)備好了數(shù)據(jù)。
這些方法的生命周期如下：對(duì)于每個(gè)分區(qū)（包含partition_id），每個(gè)微批（包含epoch_id）。
open(partitionId, epochId)方法被調(diào)用，如果open()方法返回true，則對(duì)于分區(qū)和微批中的每一行將調(diào)用process()，然后調(diào)用close(error)，在處理時(shí)拋出錯(cuò)誤（如果有）。
close()方法被調(diào)用（如果有）如果open()方法被調(diào)用并返回成功（不管返回true還是false），除非JVM或python進(jìn)程崩潰。
spark不保證輸出相同，因此無(wú)法使用（partitionId, epochId）實(shí)現(xiàn)重復(fù)數(shù)據(jù)消除，如果需要對(duì)輸出執(zhí)行重復(fù)數(shù)據(jù)消除，請(qǐng)嘗試使用foreachBatch。
自spark3.1,你也可以使用DataStreamReader.table()讀取表數(shù)據(jù)作為streaming DataFrame，使用DataStreamWriter.toTable()寫入為表。
文章版權(quán)歸作者所有，未經(jīng)允許請(qǐng)勿轉(zhuǎn)載,若此文章存在違規(guī)行為，您可以聯(lián)系管理員刪除。
轉(zhuǎn)載請(qǐng)注明本文地址：http://systransis.cn/yun/129473.html
摘要：所以，在設(shè)計(jì)時(shí)，要了解區(qū)塊鏈相關(guān)知識(shí)，這些是出于安全考慮。通過(guò)一個(gè)智能合約能夠管理所有模塊，這個(gè)是不變的，相當(dāng)于一個(gè)不變的點(diǎn)，用來(lái)鏈接各個(gè)模塊，保證穩(wěn)定，相當(dāng)于在區(qū)塊鏈上一直會(huì)有一個(gè)穩(wěn)定的地址長(zhǎng)期進(jìn)行服務(wù)。 智能合約的設(shè)計(jì)和傳統(tǒng)的應(yīng)用設(shè)計(jì)有點(diǎn)不同。傳統(tǒng)應(yīng)用一般為了快速迭代是在產(chǎn)品之后考慮安全，但是 DApp 則需要在產(chǎn)品出來(lái)之前就考慮安全問(wèn)題，它將會(huì)關(guān)系到賬戶資產(chǎn)、用戶數(shù)據(jù)等問(wèn)題，而且...
摘要：棧底是固定的，而棧頂浮動(dòng)的如果棧中元素個(gè)數(shù)為零則被稱為空棧。入棧將數(shù)據(jù)保存到棧頂。鏈棧鏈棧是指棧的鏈?zhǔn)酱鎯?chǔ)結(jié)構(gòu)，是沒(méi)有附加頭節(jié)點(diǎn)的運(yùn)算受限的單鏈表，棧頂指針是鏈表的頭指針。 一、喜歡單挑線性表 1.線性表的特性 線性表是一個(gè)線性結(jié)構(gòu)，它是一個(gè)含有n≥0個(gè)節(jié)點(diǎn)的有限序列。在節(jié)點(diǎn)中，有且僅有一個(gè)開(kāi)始節(jié)點(diǎn)沒(méi)有前驅(qū)并有一個(gè)后繼節(jié)點(diǎn)，有且僅有一個(gè)終端節(jié)點(diǎn)沒(méi)有后繼并有一個(gè)前驅(qū)節(jié)點(diǎn)。其他的節(jié)點(diǎn)都有且...
摘要：發(fā)送請(qǐng)求，處理數(shù)據(jù)。在上面這個(gè)場(chǎng)景中，這類數(shù)據(jù)的結(jié)構(gòu)可能是最常碰到的。整個(gè)過(guò)程可以簡(jiǎn)化成數(shù)據(jù)的變化引起視圖的變化，和現(xiàn)在很多前端框架數(shù)據(jù)驅(qū)動(dòng)思想有幾分相似。至此一個(gè)對(duì)于頁(yè)面的抽象出來(lái)的數(shù)據(jù)結(jié)構(gòu)雛形基本完成了。 作者：周周（滬江資深Web前端開(kāi)發(fā)工程師）本文為原創(chuàng)文章，轉(zhuǎn)載請(qǐng)注明作者及出處 前言 近期在小D十周年活動(dòng)之際，又看到了一個(gè)自家H5專題夢(mèng)工廠生成的頁(yè)面。 showImg(htt...
閱讀 1357·2023-01-11 13:20
閱讀 1707·2023-01-11 13:20
閱讀 1215·2023-01-11 13:20
閱讀 1906·2023-01-11 13:20
閱讀 4165·2023-01-11 13:20
閱讀 2757·2023-01-11 13:20
閱讀 1402·2023-01-11 13:20
閱讀 3672·2023-01-11 13:20
