之前一直不是非常理解Spark的緩存應該如何使用. 今天在使用的時候, 為了提高性能, 嘗試使用了一下Cache, 并收到了明顯的效果。

關于Cache的一些理論介紹, 網上已經很多了. 但是貌似也沒有一個簡單的例子說明。

注：因為使用的是內部數據文件, 在這邊就不公布出來了. 大家看看測試代碼跟測試結果即可。

這次測試是在JupyterNotebook這種交互式的環境下測試的. 如果是直接的submit一個job, 可能結果不太一樣。

[[219983]]

測試步驟

初始化Spark

from pyspark.sql import SparkSession  
spark = SparkSession\  
.builder\  
.appName("Cache Demo")\  
.master("spark://10.206.132.113:7077") \  
.config('spark.driver.memory', '5g') \  
.config('spark.executor.memory', '5g') \  
.config("spark.cores.max", 20) \  
.getOrCreate() 

分別讀兩個文件做測試, 并且其中一個使用Cache

ds1 = spark.read.json(os.path.join(data_path, "data.2018-01-04"))  
ds2 = spark.read.json(os.path.join(data_path, "data.2018-01-05"))  
ds1.cache() # 對于***個dataframe進行cache. 

注: 這兩個數據文件分別是1月4日跟1月5日產生的. 大小非常接近, 都是3.1G.

為了防止Spark自己做了什么Cache影響實驗, 在這里讀取兩個不同的數據文件.

計算時間：

import time  
def calc_timing(ds, app_name) :  
t1 = time.time()  
related = ds.filter("app_name = '%s'" % app_name)  
_1stRow = related.first()  
t2 = time.time()  
print "cost time:", t2 - t1 

測試結果：

calc_timing(ds1, "DrUnzip") # cost time: 13.3130679131  
calc_timing(ds2, "DrUnzip") # cost time: 18.0472488403  
calc_timing(ds1, "DrUnzip") # cost time: 0.868658065796  
calc_timing(ds2, "DrUnzip") # cost time: 15.8150720596 

可以看到:

• 對于DS1, 雖然調用了Cache ,但是因為沒有真正的使用到, 所以***次進行filter操作還是很慢的
• 第二次使用DS1的時候, 因為有了緩存, 速度快了很多
• 相對的, DS2兩次執行時間差別不大
• 如果進到Spark UI 查看具體每個Job的執行時間, 會發現, 只讀取數據文件消耗的時間也就在15~20s.

因此可以猜想, Spark的DataFrame讀取數據之后, 即使進行兩個相同的操作, 消耗的時間也不能減少, 因為Spark 默認不會把DS放到內存之中.