sparkSQL中cache的若干問題

摘要

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

場景描述

當(dāng)我們通過spark進(jìn)行統(tǒng)計和處理數(shù)據(jù)時，發(fā)現(xiàn)他是延遲計算的，如果一個應(yīng)用中出現(xiàn)多個action,而這多個action處理同一個數(shù)據(jù)源數(shù)據(jù)時，數(shù)據(jù)源用時間來過濾數(shù)據(jù)時，由于有多個action操作，遇到每個action就是一個job，每一個action都會執(zhí)行數(shù)據(jù)源獲取數(shù)據(jù)的操作，由于兩個action之間的操作存在時間差，這兩個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")
//第一個Action操作
val spark.sql("""
insert overwrite table default.test2
*
from data1
""")

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

如果在運(yùn)行第二個Action操作前，test1表中又增加了一條記錄3,2018-07-01 13：12：04
即執(zhí)行第一個Action時記錄還是兩條1和2，而再執(zhí)行完第一個Action后而又執(zhí)行第二個Action之前，
增加了一個新的單子：3,2018-07-01 13：12：04
那么在test2表中的數(shù)據(jù)是怎么樣的呢？
第一種情況（因為第二個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í)行計劃還是不太熟悉。首先spark是lazy計算的，即不觸發(fā)action操作，其實不提交作業(yè)的。而在這個application中存在兩個action，而這兩個aciton使用了同一個數(shù)據(jù)源的rdd，應(yīng)該稱為變量odsData，當(dāng)遇到第一個action，其會把自己這個執(zhí)行鏈上的rdd都執(zhí)行一遍，包括執(zhí)行odsData，而遇到第二個aciton的時候，其也會把自己的執(zhí)行鏈上的數(shù)據(jù)又執(zhí)行了一遍包括odsData，并從數(shù)據(jù)源中重新取數(shù)。有人會疑惑，第一個action在執(zhí)行的時候，已經(jīng)執(zhí)行了odsData，這個RDD的結(jié)果不應(yīng)該緩存起來嗎？個人認(rèn)為，spark還沒有那么的智能，并且網(wǎng)上經(jīng)常說的job，stage，rdd，task的劃分應(yīng)該是在同一個job內(nèi)進(jìn)行的。而同一個應(yīng)用中夸job的stage拆分是不存在的。那么出現(xiàn)這個結(jié)果應(yīng)該怎么辦呢？
cache的出場

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

//第一個Action
spark.sql(
"""
|INSERT OVERWRITE TABLE default.test1
|SELECT
|
|FROM dwData
""".stripMargin)
//改變數(shù)據(jù)源表test表的數(shù)據(jù)并且是第二個Action
spark.sql(
"""
|INSERT OVERWRITE TABLE default.test
|SELECT
| 1,
| "2017",
| "2018",
| "2018"
|FROM default.test
""".stripMargin)
//第三個Action和第一個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根本就沒有起到效果，或者說第三個Action根本就沒有使用我們cache的數(shù)據(jù)。這次我把日志都打出來了啊。
第一個Action的聲明周期：

第三個Action的日志：

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

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

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