摘要

sparkSQL在使用cache緩存的時(shí)候，有時(shí)候緩存可能不起作用，可能會(huì)發(fā)出緩存是假的吧的感慨?，F(xiàn)在我們就把這個(gè)問題說道說道。
問題

為焉耆等地區(qū)用戶提供了全套網(wǎng)頁設(shè)計(jì)制作服務(wù)，及焉耆網(wǎng)站建設(shè)行業(yè)解決方案。主營業(yè)務(wù)為做網(wǎng)站、成都網(wǎng)站制作、焉耆網(wǎng)站設(shè)計(jì)，以傳統(tǒng)方式定制建設(shè)網(wǎng)站，并提供域名空間備案等一條龍服務(wù)，秉承以專業(yè)、用心的態(tài)度為用戶提供真誠的服務(wù)。我們深信只要達(dá)到每一位用戶的要求，就會(huì)得到認(rèn)可，從而選擇與我們長期合作。這樣，我們也可以走得更遠(yuǎn)！

場景描述

當(dāng)我們通過spark進(jìn)行統(tǒng)計(jì)和處理數(shù)據(jù)時(shí)，發(fā)現(xiàn)他是延遲計(jì)算的，如果一個(gè)應(yīng)用中出現(xiàn)多個(gè)action,而這多個(gè)action處理同一個(gè)數(shù)據(jù)源數(shù)據(jù)時(shí)，數(shù)據(jù)源用時(shí)間來過濾數(shù)據(jù)時(shí)，由于有多個(gè)action操作，遇到每個(gè)action就是一個(gè)job，每一個(gè)action都會(huì)執(zhí)行數(shù)據(jù)源獲取數(shù)據(jù)的操作，由于兩個(gè)action之間的操作存在時(shí)間差，這兩個(gè)action獲取的數(shù)據(jù)有可能不一致。
例如下例
test1表中的數(shù)據(jù)

1，2018-07-01 10：10：03
2，2018-07-01 11：12：04

代碼如下操作
val odsData = spark.sql("""
select
from default.test1
where time < "2018-07-02"
""")
val targetData = odsData.map(fun _)
val targetData.createOrReplaceTempView("data1")
//第一個(gè)Action操作
val spark.sql("""
insert overwrite table default.test2
*
from data1
""")

val targetData1 = odsData.map(fun2 _) //引用同一個(gè)數(shù)據(jù)源
targetData1.createOrReplaceTempView("data2")
//第二個(gè)action操作
val spark.sql("""
insert table default.test2
*
from data2
""")

如果在運(yùn)行第二個(gè)Action操作前，test1表中又增加了一條記錄3,2018-07-01 13：12：04
即執(zhí)行第一個(gè)Action時(shí)記錄還是兩條1和2，而再執(zhí)行完第一個(gè)Action后而又執(zhí)行第二個(gè)Action之前，
增加了一個(gè)新的單子：3,2018-07-01 13：12：04
那么在test2表中的數(shù)據(jù)是怎么樣的呢？
第一種情況（因?yàn)榈诙€(gè)action是insert而不是insert overwrite）

1，2018-07-01 10：10：03
2，2018-07-01 11：12：04
1，2018-07-01 10：10：03
2，2018-07-01 11：12：04

第二種情況

1，2018-07-01 10：10：03
2，2018-07-01 11：12：04
1，2018-07-01 10：10：03
2，2018-07-01 11：12：04
3,2018-07-01 13：12：04

結(jié)果分析

結(jié)果是第二中情況。如果認(rèn)為是第一種情況的對spark的執(zhí)行計(jì)劃還是不太熟悉。首先spark是lazy計(jì)算的，即不觸發(fā)action操作，其實(shí)不提交作業(yè)的。而在這個(gè)application中存在兩個(gè)action，而這兩個(gè)aciton使用了同一個(gè)數(shù)據(jù)源的rdd，應(yīng)該稱為變量odsData，當(dāng)遇到第一個(gè)action，其會(huì)把自己這個(gè)執(zhí)行鏈上的rdd都執(zhí)行一遍，包括執(zhí)行odsData，而遇到第二個(gè)aciton的時(shí)候，其也會(huì)把自己的執(zhí)行鏈上的數(shù)據(jù)又執(zhí)行了一遍包括odsData，并從數(shù)據(jù)源中重新取數(shù)。有人會(huì)疑惑，第一個(gè)action在執(zhí)行的時(shí)候，已經(jīng)執(zhí)行了odsData，這個(gè)RDD的結(jié)果不應(yīng)該緩存起來嗎？個(gè)人認(rèn)為，spark還沒有那么的智能，并且網(wǎng)上經(jīng)常說的job，stage，rdd，task的劃分應(yīng)該是在同一個(gè)job內(nèi)進(jìn)行的。而同一個(gè)應(yīng)用中夸job的stage拆分是不存在的。那么出現(xiàn)這個(gè)結(jié)果應(yīng)該怎么辦呢？
cache的出場

當(dāng)出現(xiàn)這樣的情況時(shí)，我的應(yīng)用每天就會(huì)漏幾十條數(shù)據(jù)，很是煩人，最后發(fā)現(xiàn)了上面的問題，當(dāng)時(shí)想解決方案時(shí)，第一個(gè)就是想到了cache，我把第一次執(zhí)行Action操作時(shí)，把odsData給緩存了，這樣應(yīng)該不會(huì)有什么問題了吧。從而可以保證兩個(gè)action操作，同一個(gè)數(shù)據(jù)源的數(shù)據(jù)一致性。只能說too young to sample了。這樣解決不了上面出現(xiàn)的問題。同樣以一個(gè)例子來看。
test表中的數(shù)據(jù)：

1 2017-01-01 01:00:00 2016-05-04 9999-12-31
2 2017-01-01 02:00:00 2016-01-01 9999-12-31
代碼：

val curentData = spark.sql(
"""
|select
|*
|from default.test
""".stripMargin)

curentData.cache() //緩存我們的結(jié)果

curentData.createOrReplaceTempView("dwData")

//第一個(gè)Action
spark.sql(
"""
|INSERT OVERWRITE TABLE default.test1
|SELECT
|
|FROM dwData
""".stripMargin)
//改變數(shù)據(jù)源表test表的數(shù)據(jù)并且是第二個(gè)Action
spark.sql(
"""
|INSERT OVERWRITE TABLE default.test
|SELECT
| 1,
| "2017",
| "2018",
| "2018"
|FROM default.test
""".stripMargin)
//第三個(gè)Action和第一個(gè)Action同數(shù)據(jù)源,并且cache第一次運(yùn)行的結(jié)果。
spark.sql(
"""
|INSERT OVERWRITE TABLE default.test1
|SELECT
|
|FROM dwData
""".stripMargin)
那么test1表中的結(jié)果
第一種情況：
1 2017-01-01 01:00:00 2016-05-04 9999-12-31
2 2017-01-01 02:00:00 2016-01-01 9999-12-31
第二種情況

1 2017 2018 2018
1 2017 2018 2018
結(jié)果分析

結(jié)果是第二種情況，也就是說我們cache根本就沒有起到效果，或者說第三個(gè)Action根本就沒有使用我們cache的數(shù)據(jù)。這次我把日志都打出來了啊。
第一個(gè)Action的聲明周期：

第三個(gè)Action的日志：

從這兩個(gè)日志可以看出，我們設(shè)置cache其只能在同一個(gè)job中生效。而夸job的使用這樣的數(shù)據(jù)緩存數(shù)據(jù)是不存在的。
如果想更加詳細(xì)的了解cache的原理和作用，可以去網(wǎng)上搜，大把大把的資料，但是一定要記住，網(wǎng)上說的要限定一個(gè)條件，在同一個(gè)job內(nèi)的rdd，夸job的cache是不存在的。
解決方案

我們最終希望解決的事，當(dāng)兩個(gè)action想要使用同一個(gè)數(shù)據(jù)源的rdd的時(shí)候，如何保證其數(shù)據(jù)的一致性。
方案：
把第一個(gè)Action算子用到的數(shù)據(jù)源給寫入到一個(gè)臨時(shí)表中
然后再第二個(gè)Action中，直接讀取臨時(shí)表的數(shù)據(jù)，而不是直接使用odsData
更好的方案還沒有想好，可以根據(jù)業(yè)務(wù)的不同來搞。

第二個(gè)方案現(xiàn)在就是我們使用spark提供的checkpoint機(jī)制，checkpoint會(huì)把我們的數(shù)據(jù)
自動(dòng)緩存到hdfs，它就會(huì)把這個(gè)rdd以前的父rdd的數(shù)據(jù)全部刪除，以后不管哪個(gè)job的rdd
需要使用這個(gè)rdd的數(shù)據(jù)時(shí)，都會(huì)從這個(gè)checkpoin的目錄中讀取數(shù)據(jù)。
spark.sparkContext.setCheckpointDir("hdfs://hadoop-1:5000/hanfangfang")
curentData.cache().checkpoint
這樣就可以使不同的job，同一個(gè)數(shù)據(jù)源數(shù)據(jù)的一致性。
同時(shí)我們也要記住，當(dāng)程序運(yùn)行完成，其不會(huì)刪除checkpoint的數(shù)據(jù)的，需要們手動(dòng)刪除。

另外有需要云服務(wù)器可以了解下創(chuàng)新互聯(lián)scvps.cn，海內(nèi)外云服務(wù)器15元起步，三天無理由+7*72小時(shí)售后在線，公司持有idc許可證，提供“云服務(wù)器、裸金屬服務(wù)器、高防服務(wù)器、香港服務(wù)器、美國服務(wù)器、虛擬主機(jī)、免備案服務(wù)器”等云主機(jī)租用服務(wù)以及企業(yè)上云的綜合解決方案，具有“安全穩(wěn)定、簡單易用、服務(wù)可用性高、性價(jià)比高”等特點(diǎn)與優(yōu)勢，專為企業(yè)上云打造定制，能夠滿足用戶豐富、多元化的應(yīng)用場景需求。

網(wǎng)站標(biāo)題：sparkSQL中cache的若干問題-創(chuàng)新互聯(lián)
分享地址：http://jinyejixie.com/article6/piiig.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供品牌網(wǎng)站制作、標(biāo)簽優(yōu)化、商城網(wǎng)站、虛擬主機(jī)、網(wǎng)站建設(shè)、App設(shè)計(jì)

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來源： 創(chuàng)新互聯(lián)