流批一體的近實(shí)時(shí)數(shù)倉(cāng)的思考與設(shè)計(jì)

流批一體的近實(shí)時(shí)數(shù)倉(cāng)的思考與設(shè)計(jì)導(dǎo)讀：本文基于對(duì)數(shù)據(jù)時(shí)間旅行的思考，引出了對(duì)目前三種數(shù)倉(cāng)形態(tài)和兩種數(shù)倉(cāng)架構(gòu)的思考。結(jié)合數(shù)據(jù)湖在Flink的應(yīng)用和數(shù)據(jù)湖元數(shù)據(jù)類型的思考，探索了基于數(shù)據(jù)湖的FlinkSQL流批一體的實(shí)踐，在流批一體SQL表達(dá)一致、結(jié)果一致性、流批任務(wù)分離、混合調(diào)度依賴等進(jìn)行了設(shè)計(jì)和探索。01數(shù)據(jù)的時(shí)間旅行和業(yè)務(wù)對(duì)數(shù)據(jù)的本質(zhì)要求大規(guī)模的數(shù)據(jù)處理興起于Hadoop生態(tài)的發(fā)展，關(guān)鍵在于分布式儲(chǔ)存和分布式計(jì)算的發(fā)展，造就了如今近百種有關(guān)大數(shù)據(jù)的生態(tài)技術(shù)。數(shù)倉(cāng)理論和建模理論基于大數(shù)據(jù)技術(shù)體系得以快速發(fā)展，其中離線數(shù)倉(cāng)的標(biāo)準(zhǔn)化建設(shè)得到了廣泛應(yīng)用。數(shù)據(jù)的本質(zhì)是一種行為的具象，業(yè)務(wù)在對(duì)數(shù)據(jù)的需求，核心在于對(duì)行為的可探索和可觀察?；诖?，我們需要明確一點(diǎn)，大數(shù)據(jù)技術(shù)是否完全滿足了業(yè)務(wù)對(duì)數(shù)據(jù)需求在時(shí)間維度上的確定性了呢，這點(diǎn)是值得思考的。那么我們先來(lái)看一下數(shù)據(jù)的時(shí)間旅行。業(yè)務(wù)期望的數(shù)據(jù)：用戶空間下的時(shí)間數(shù)據(jù)，t1時(shí)間數(shù)據(jù)，用戶自然時(shí)間點(diǎn)或自然時(shí)間段的明細(xì)或者統(tǒng)計(jì)數(shù)據(jù)。

傳輸延遲：App用戶，數(shù)據(jù)發(fā)送到網(wǎng)關(guān)或者日志服務(wù)系統(tǒng)，或者Server日志落文件系統(tǒng)所產(chǎn)生的延遲。Event進(jìn)入到存儲(chǔ)空間，可以代表數(shù)據(jù)已經(jīng)是確定的，基本可觀察，一般情況下，這個(gè)延遲很小。但是，在某些情況，比如APP的日志產(chǎn)生之后，但是因?yàn)榫W(wǎng)絡(luò)等問(wèn)題一直沒(méi)有發(fā)送，或者Server宕機(jī)，導(dǎo)致延遲發(fā)送或者最終丟失?？傮w而言，傳輸延遲屬于不可控延遲，暫時(shí)沒(méi)有什么好的技術(shù)方案來(lái)解決。

存儲(chǔ)空間：數(shù)據(jù)承載于實(shí)際的存儲(chǔ)中，離線數(shù)倉(cāng)承載于具體的分布式文件系統(tǒng)，實(shí)時(shí)數(shù)倉(cāng)基于Kafka的消息隊(duì)列系統(tǒng)，近實(shí)時(shí)數(shù)倉(cāng)承載于數(shù)據(jù)湖存儲(chǔ)中。這里可以抽象來(lái)看離線數(shù)倉(cāng)，Event承載于分布式文件系統(tǒng)，以小時(shí)分區(qū)為例，某個(gè)小時(shí)的分區(qū)本質(zhì)是自然時(shí)間產(chǎn)生的文件的集合，時(shí)間精度退化為小時(shí)級(jí)別。

計(jì)算延遲：數(shù)據(jù)進(jìn)入存儲(chǔ)之后，與進(jìn)入計(jì)算空間的時(shí)間差，t3-t2。實(shí)時(shí)數(shù)倉(cāng)中，計(jì)算延遲是數(shù)據(jù)的ProcessTime-IngestTime。離線數(shù)倉(cāng)中，計(jì)算延遲是調(diào)度產(chǎn)生實(shí)例運(yùn)行時(shí)間-數(shù)據(jù)進(jìn)入存儲(chǔ)空間的時(shí)間差。本質(zhì)離線數(shù)倉(cāng)和實(shí)時(shí)數(shù)倉(cāng)的計(jì)算延遲在抽象上看是一致的。計(jì)算延遲在不同的數(shù)倉(cāng)體系下，產(chǎn)生的時(shí)效不同，我們會(huì)劃分為三種主流的數(shù)倉(cāng)體系，秒級(jí)的實(shí)時(shí)數(shù)倉(cāng)，分鐘級(jí)的近實(shí)時(shí)數(shù)倉(cāng)，小時(shí)級(jí)的離線數(shù)倉(cāng)?？梢钥闯?，數(shù)倉(cāng)的時(shí)效性差異，因?yàn)閭鬏斞舆t的不可控，退化為計(jì)算延遲的差異。02離線、近實(shí)時(shí)、實(shí)時(shí)三種數(shù)倉(cāng)在時(shí)間維度下的成因

在離線數(shù)倉(cāng)和實(shí)時(shí)數(shù)倉(cāng)，常常會(huì)提到數(shù)據(jù)的有界和無(wú)界，認(rèn)為離線數(shù)倉(cāng)的數(shù)據(jù)是有界的，實(shí)時(shí)數(shù)倉(cāng)的消息流是無(wú)界的。準(zhǔn)確與否在于數(shù)據(jù)的確定性考量。

離線數(shù)倉(cāng)的確定性，在于文件自然生成時(shí)間的確定性和不可更改性，某個(gè)小時(shí)的自然文件生成，近似等于事件時(shí)間在自然時(shí)間的確定性，反例就是我們能看到數(shù)據(jù)漂移的情況，事件時(shí)間會(huì)或多或少落入上個(gè)小時(shí)或者下個(gè)小時(shí)的自然文件生成時(shí)間。那么離線數(shù)倉(cāng)的確定性，實(shí)質(zhì)是數(shù)據(jù)的IngestTime的確定性，具有天然的文件屬性，易于分割。當(dāng)我們說(shuō)離線數(shù)倉(cāng)計(jì)算的數(shù)據(jù)是準(zhǔn)確的時(shí)候，默認(rèn)了傳輸延遲帶來(lái)的影響很小或者默認(rèn)了當(dāng)前小時(shí)的數(shù)據(jù)指標(biāo)的標(biāo)準(zhǔn)是文件的自然形成時(shí)間。實(shí)時(shí)數(shù)倉(cāng)，常常會(huì)提及不確定性或者說(shuō)Lambda架構(gòu)實(shí)際是對(duì)實(shí)時(shí)數(shù)倉(cāng)的不確定性的替代方案。這種不確定性的原因是什么呢？這里分為四類情況說(shuō)明，一是ETL的處理，從窗口上來(lái)說(shuō)，是單條數(shù)據(jù)即為一個(gè)窗口，窗口的產(chǎn)生和銷毀在一個(gè)Event中完成，y=window(data)。二是基于EventTime的時(shí)間窗口，如果再定義延遲時(shí)間，y=window(datas,datas.EventTime,delay)，第三種和第四種分別就是IngestTime和ProcessTime的時(shí)間窗口函數(shù)。對(duì)比離線數(shù)倉(cāng)，可以看出，基于IngestTime的時(shí)間窗口和離線數(shù)倉(cāng)的時(shí)間語(yǔ)義最為一致。離線數(shù)倉(cāng)在時(shí)間窗口上，可以看做為數(shù)據(jù)進(jìn)入文件的自然時(shí)間所對(duì)應(yīng)的小時(shí)窗口，數(shù)據(jù)所承載的文件的確定性，保證了小時(shí)窗口的數(shù)據(jù)確定性，y=window(files)。近實(shí)時(shí)數(shù)倉(cāng)，比如基于Iceberg的數(shù)據(jù)湖建立的近實(shí)時(shí)數(shù)倉(cāng)，在于離線數(shù)倉(cāng)對(duì)比中，實(shí)際是將基于小時(shí)文件細(xì)分到分鐘級(jí)別的快照文件上來(lái)，y=window(snapshots)。對(duì)比實(shí)時(shí)數(shù)倉(cāng)，因?yàn)镵afka的IngestTime目前在精確性上是不精確的，基于快照的文件劃分，在精確性上有一定的保證，同時(shí)在降低時(shí)效程度，從秒退化為分鐘，很多場(chǎng)景是可以容忍的。三種在時(shí)間維度對(duì)比上看，一是在某個(gè)時(shí)間，統(tǒng)計(jì)的本質(zhì)對(duì)業(yè)務(wù)的需求都是近似的，這個(gè)本質(zhì)是傳輸延遲所帶來(lái)的，但是這個(gè)在實(shí)踐中，不影響數(shù)據(jù)的可用性和統(tǒng)計(jì)學(xué)意義。二是不同數(shù)倉(cāng)的劃分，是存儲(chǔ)和計(jì)算技術(shù)發(fā)展所帶來(lái)的。三是離線數(shù)倉(cāng)的確定性模糊了傳輸延遲，實(shí)時(shí)數(shù)倉(cāng)的不確定性，是對(duì)傳輸延遲的一種取舍，人為的限定了EventTime的最大延遲時(shí)間，從而保證了指標(biāo)的時(shí)效性，都是具有實(shí)踐的意義所在。03Lambda和Kappa架構(gòu)在時(shí)間維度下的取舍

當(dāng)離線數(shù)倉(cāng)剛剛發(fā)展的時(shí)候，只有一種數(shù)倉(cāng)架構(gòu)，也是基于大數(shù)據(jù)分布式處理剛剛發(fā)展的原因。隨著實(shí)時(shí)技術(shù)的發(fā)展，大家在時(shí)效性上有了更多要求，但是同離線數(shù)倉(cāng)對(duì)比的時(shí)候，在數(shù)據(jù)的準(zhǔn)確性上，因?yàn)榻y(tǒng)計(jì)的窗口不同，必然會(huì)導(dǎo)致某個(gè)時(shí)刻的指標(biāo)結(jié)果的不嚴(yán)格一致。

為了解決這種不嚴(yán)格一致的情況，Lambda架構(gòu)（由Storm的作者NathanMarz提出的）產(chǎn)生的，實(shí)時(shí)確保時(shí)效，離線確保準(zhǔn)確。最終會(huì)以確保離線三個(gè)時(shí)間窗口的統(tǒng)計(jì)一個(gè)事件時(shí)間窗口的結(jié)果，來(lái)回補(bǔ)實(shí)時(shí)數(shù)倉(cāng)以為EventTime窗口，因?yàn)闀r(shí)效性丟棄的延遲數(shù)據(jù)的結(jié)果，從而保證業(yè)務(wù)上對(duì)EventTime窗口的要求，或者默認(rèn)為離線的IngestTime所產(chǎn)生的文件分區(qū)近似認(rèn)為EventTime的時(shí)間窗口。這種帶來(lái)的弊端，維護(hù)兩套數(shù)據(jù)路線，而大家總在想辦法解決。

Kappa架構(gòu)的提出，得益于實(shí)時(shí)計(jì)算的效率提升，但是因?yàn)樵谂幚砑夹g(shù)短板，生產(chǎn)實(shí)踐推廣受限。Kappa架構(gòu)是基于實(shí)時(shí)EventTime的一種數(shù)據(jù)窗口處理，因?yàn)镵afka的IngestTime不精確和為了同離線數(shù)倉(cāng)對(duì)比而權(quán)衡考慮，EventTime在傳輸延遲上的不可控，導(dǎo)致Kappa架構(gòu)的準(zhǔn)確性就會(huì)出現(xiàn)折扣。雖然是業(yè)務(wù)上最準(zhǔn)確的時(shí)間范圍，可行性上確不佳。

近些年來(lái)，不斷發(fā)展的MPP架構(gòu)的OLAP查詢引擎，并不會(huì)涉及到時(shí)間窗口的計(jì)算取舍，OLAP引擎本質(zhì)是基于ProcessTime來(lái)加速查詢的一種技術(shù)手段，是數(shù)倉(cāng)不可分割的一部分，但是傳輸延遲的不可控沒(méi)有解決，但是將計(jì)算延遲下推到了查詢時(shí)，通過(guò)快速查詢來(lái)解決盡可能減少計(jì)算延遲，同時(shí)保證了查詢的靈活性，自助分析探索上有著廣泛的應(yīng)用。

從數(shù)倉(cāng)架構(gòu)的發(fā)展上看，不斷在圍繞結(jié)果的確定性，技術(shù)的可行性，數(shù)據(jù)的時(shí)效性，查詢的靈活性上，不斷的權(quán)衡，各個(gè)組件也是依據(jù)實(shí)際需求而發(fā)展起來(lái)的。04數(shù)倉(cāng)一體的可行性思考

基于三種數(shù)倉(cāng)體系和兩種架構(gòu)的思考，每個(gè)設(shè)計(jì)都是兼顧一種或多種考量，那么能不能實(shí)現(xiàn)一種機(jī)制，能夠較好的滿足數(shù)倉(cāng)需求體系建設(shè)呢？從目前的技術(shù)發(fā)展上看，是有一定的可能性的。架構(gòu)體系的發(fā)展一是基于技術(shù)基礎(chǔ)，二是不斷吸收組件的優(yōu)點(diǎn)，做加法。

除去實(shí)時(shí)、近實(shí)時(shí)、離線數(shù)倉(cāng)的劃分，從技術(shù)的視角去看數(shù)倉(cāng)建設(shè)的可行性。那么我們就要選取一些重要的點(diǎn)，取舍掉一些不可能的實(shí)現(xiàn)。

第一點(diǎn)是結(jié)果的確定性，這點(diǎn)是基于離線數(shù)倉(cāng)發(fā)展的思考。不確定性帶來(lái)的問(wèn)題是信息的不對(duì)稱，確定性的結(jié)果是可以模糊一定的指標(biāo)含義的。

第二點(diǎn)是數(shù)據(jù)的時(shí)效性，高時(shí)效必然能夠滿足低時(shí)效，反之不然。另外數(shù)據(jù)的時(shí)效性，本身是基礎(chǔ)組件的技術(shù)發(fā)展所限制的。

第三點(diǎn)是開發(fā)的便利性，排在時(shí)效性后面的考慮是，便利性是基于應(yīng)用層面建設(shè)的，難度一般是弱于基礎(chǔ)組件的，可以通過(guò)不斷實(shí)踐優(yōu)化，達(dá)到一個(gè)良好的使用體驗(yàn)。第四點(diǎn)是查詢的靈活性和高響應(yīng)，OLAP的基礎(chǔ)設(shè)計(jì)保證了查詢速度，那么OLAP的技術(shù)架構(gòu)體現(xiàn)是可以復(fù)用或者拓展的。

那么基于上面四點(diǎn)考慮，可以在實(shí)時(shí)數(shù)倉(cāng)的基礎(chǔ)上，優(yōu)先解決掉確定性問(wèn)題。這個(gè)是很重要的一個(gè)命題，要保證計(jì)算結(jié)果同離線數(shù)倉(cāng)的一致性。這一點(diǎn)的實(shí)現(xiàn)方面，可以參考離線數(shù)倉(cāng)，模糊EventTime和IngestTime，用文件的start和end作為確定性的依據(jù)，文件的中間實(shí)時(shí)計(jì)算，確保時(shí)效性。那么基于Flink，就需要實(shí)現(xiàn)一種基于文件自然分割的Watermark機(jī)制，作為計(jì)算窗口劃分的依據(jù)。

在確定性問(wèn)題之后，需要解決計(jì)算的成本和使用的成本，這里比較重要的是存儲(chǔ)層，實(shí)時(shí)數(shù)倉(cāng)依賴Kafka，Kafka發(fā)展不具備數(shù)倉(cāng)一些重要的點(diǎn)，成本是一個(gè)方面，查詢是一個(gè)方面，Kafka無(wú)法架構(gòu)在各種OLAP引擎或者計(jì)算引擎上面。這里，近實(shí)時(shí)數(shù)倉(cāng)的依賴，比如數(shù)據(jù)湖或者Paimon，數(shù)據(jù)湖分鐘級(jí)的時(shí)效。不過(guò)，從發(fā)展的角度上看，是一種可行的解決方案。數(shù)據(jù)湖兼顧了流計(jì)算和批計(jì)算，同時(shí)，如果未來(lái)OLAP引擎如果能夠在數(shù)據(jù)湖上實(shí)現(xiàn)類似MPP架構(gòu)的查詢效率，這也是有可能的，比如短期可以用數(shù)據(jù)冗余，將數(shù)據(jù)湖格式的數(shù)據(jù)轉(zhuǎn)換一份到OLAP對(duì)應(yīng)的引擎上實(shí)現(xiàn)加速查詢。

第三個(gè)方面，流式計(jì)算的管理和依賴機(jī)制，借鑒于離線數(shù)倉(cāng)的管理方式，需要一套完備的數(shù)據(jù)依賴管理，任務(wù)容錯(cuò)回跑機(jī)制。實(shí)時(shí)數(shù)倉(cāng)一般是基于單個(gè)任務(wù)式的管理，離線數(shù)倉(cāng)是基于任務(wù)流的管理，那么實(shí)時(shí)數(shù)倉(cāng)的發(fā)展，也必然要實(shí)現(xiàn)任務(wù)流的管理方式，覆蓋整個(gè)開發(fā)鏈路。

為了實(shí)現(xiàn)一種統(tǒng)計(jì)的數(shù)倉(cāng)架構(gòu)，那么需要的發(fā)展工作如下：一是著重發(fā)展存儲(chǔ)層，比如數(shù)據(jù)湖，既要比較好的適應(yīng)流和批引擎，又要能夠高度適應(yīng)OLAP查詢引擎。二是在實(shí)時(shí)數(shù)倉(cāng)或者近實(shí)時(shí)數(shù)倉(cāng)，引入類似離線數(shù)倉(cāng)的調(diào)度依賴管理和補(bǔ)數(shù)和容錯(cuò)回跑機(jī)制，或者在離線調(diào)度上兼容流任務(wù)依賴調(diào)度，實(shí)現(xiàn)任務(wù)流級(jí)別的管理和流批一體的數(shù)倉(cāng)實(shí)現(xiàn)。三是在引擎層著重發(fā)展Flink批處理能力。最終的任務(wù)運(yùn)行方式同時(shí)包含三種：實(shí)時(shí)模式、離線模式、業(yè)務(wù)模式，分別對(duì)應(yīng)著不同的數(shù)據(jù)準(zhǔn)確性級(jí)別。也可以任選其一或者其二作為運(yùn)行方式。05基于Flink和數(shù)據(jù)湖的流批一體近實(shí)時(shí)數(shù)倉(cāng)設(shè)計(jì)示例

數(shù)倉(cāng)任務(wù)在離線調(diào)度和實(shí)時(shí)任務(wù)的簡(jiǎn)單抽象示例:數(shù)據(jù)源=>同步任務(wù)/實(shí)時(shí)任務(wù)=>stg_table(partition=hour)=>計(jì)算任務(wù)(insertoverwritepartition=hour)=>dwd_table(partition=hour)=>計(jì)算任務(wù)(insertoverwritepartition=hour)=>dws_table(partition=hour)=>同步任務(wù)=>OLAP加速=>數(shù)據(jù)服務(wù)

如果存儲(chǔ)層是基于數(shù)據(jù)湖（以Paimon為例）：離線調(diào)度產(chǎn)生的表的版本信息，commit_kind:insertoverwrite類型的。同時(shí)離線任務(wù)的驅(qū)動(dòng)，是基于調(diào)度依賴的驅(qū)動(dòng)，onebyone的調(diào)度。

如果是基于流式計(jì)算，比如分鐘級(jí)生成snapshot那么會(huì)演變?yōu)椋簲?shù)據(jù)源=>同步任務(wù)/實(shí)時(shí)任務(wù)=>stg_table(version=snapshot_id)=>計(jì)算任務(wù)(insertintoversion=snapshot_id)=>dwd_table(version=snapshot_id)=>計(jì)算任務(wù)(insertintoversion=snapshot_id)=>dws_table(version=snapshot_id)=>同步任務(wù)=>OLAP加速=>數(shù)據(jù)服務(wù)那么啟動(dòng)多個(gè)任務(wù)，任務(wù)是持續(xù)的運(yùn)行。commit_kind:insertinto類型的。

那么要想實(shí)現(xiàn)流批一體的近實(shí)時(shí)數(shù)倉(cāng)，需要解決如下問(wèn)題：

1.Flink任務(wù)支持批量計(jì)算能力要持續(xù)不斷的加強(qiáng)

從Flink1.16/1.17的版本發(fā)布情況，在批處理能力上有比較大的提升，同時(shí)，社區(qū)也在持續(xù)不斷的加強(qiáng)批處理能力以及同hive的兼容能力。

2.如何使用同一份FlinkSQL，既可以用于批任務(wù)調(diào)度，又可以用于流任務(wù)運(yùn)行呢兩張表：dwd_partition_word_count，dws_partition_word_count，計(jì)算wordcountCREATETABLEtablestore.tablestore_test.dwd_partition_word_count(logdateString,user_idbigint)PARTITIONEDBY(logdate)WITH('bucket'='3');

CREATETABLEtablestore.tablestore_test.dws_partition_word_count(logdateString,user_idbigint,cntBIGINT,PRIMARYKEY(logdate,user_id)NOTENFORCED)PARTITIONEDBY(logdate)WITH('bucket'='3');批任務(wù)的FlinkSQL：insertoverwritetablestore.tablestore_test.dws_partition_word_countPARTITION(logdate=${start_date})selectuser_id,count(1)ascntfromtablestore.tablestore_test.dwd_partition_word_countwherelogdate=${start_date}groupbyuser_id;--或者insertoverwritetablestore.tablestore_test.dws_partition_word_countselectlogdate,user_id,count(1)ascntfromtablestore.tablestore_test.dwd_partition_word_countwherelogdate=${start_date}groupbylogdate,user_id;流任務(wù)的FlinkSQL:insertintotablestore.tablestore_test.dws_partition_word_countselectlogdate,user_id,count(1)ascntfromtablestore.tablestore_test.dwd_partition_word_countgroupbylogdate,user_id;

如何用一個(gè)FlinkSQL來(lái)實(shí)現(xiàn)流批模型下的不同呢？

不同點(diǎn)：Insertinto和Insertoverwrite的問(wèn)題，這個(gè)通過(guò)在提交運(yùn)行模式的時(shí)候，如果是批任務(wù)，則是InsertOverwrite，如果是流任務(wù)，則轉(zhuǎn)為Insertinto，這個(gè)在技術(shù)上沒(méi)有什么難點(diǎn)。

不同點(diǎn)：Where條件的數(shù)據(jù)范圍問(wèn)題。抽象來(lái)看，流任務(wù)和批任務(wù)的時(shí)間范圍在表達(dá)上是可以統(tǒng)一的insertoverwritetablestore.tablestore_test.dws_partition_word_countselectlogdate,user_id,count(1)ascntfromtablestore.tablestore_test.dwd_partition_word_countwherelogdate>=${start_date}andlogdate<=${end_date}groupbylogdate,user_id;

比如跑4月22號(hào)一天的數(shù)據(jù)，執(zhí)行的批SQL為：insertoverwritetablestore.tablestore_test.dws_partition_word_countselect

logdate,

user_id,count(1)

as

cnt

from

tablestore.tablestore_test.dwd_partition_word_count

where

logdate>='20230422'

and

logdate<='20230422'

group

by

logdate,user_id;

如果用流模式跑，執(zhí)行的SQL可以為：select

logdate,

user_id,count(1)

as

cnt

from

tablestore.tablestore_test.dwd_partition_word_count

where

logdate>='19700101'

and

logdate<='99990101'

group

by

logdate,user_id;

insertoverwrite/into和時(shí)間范圍，可以由平臺(tái)執(zhí)行的時(shí)候自動(dòng)轉(zhuǎn)換和參數(shù)輸入。

3.批任務(wù)的調(diào)度和流任務(wù)的計(jì)算如何分離

任務(wù)完成開發(fā)，在批模式下，用調(diào)度任務(wù)驗(yàn)證了邏輯無(wú)誤，那么之后可以用流模式，一直持續(xù)不斷的運(yùn)行。一是計(jì)算邏輯變更或者歷史數(shù)據(jù)修復(fù)怎么辦，二是可不可以支持流批雙跑。其實(shí)本質(zhì)是一個(gè)問(wèn)題。如果計(jì)算邏輯變更，那么可以修改流批一體的SQL邏輯，然后流任務(wù)