之前一直不是非常理解 Spark 的缓存应该如何使用. 今天在使用的时候, 为了提高性能, 尝试使用了一下 Cache, 并收到了明显的效果.
关于 Cache 的一些理论介绍, 网上已经很多了. 但是貌似也没有一个简单的例子说明.
注:
因为使用的是内部数据文件, 在这边就不公布出来了. 大家看看测试代码跟测试结果即可.
这次测试是在 JupyterNotebook 这种交互式的环境下测试的. 如果是直接的 submit 一个 job, 可能结果不太一样.
测试步骤
初始化 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.3130679131calc_timing(ds2, "DrUnzip")#cost time: 18.0472488403calc_timing(ds1, "DrUnzip")#cost time: 0.868658065796calc_timing(ds2, "DrUnzip")#cost time: 15.8150720596
可以看到:
对于 DS1, 虽然调用了 Cache , 但是因为没有真正的使用到, 所以第一次进行 filter 操作还是很慢的
第二次使用 DS1 的时候, 因为有了缓存, 速度快了很多
相对的, DS2 两次执行时间差别不大
如果进到 Spark UI 查看具体每个 Job 的执行时间, 会发现, 只读取数据文件消耗的时间也就在 15~20s.
因此可以猜想, Spark 的 DataFrame 读取数据之后, 即使进行两个相同的操作, 消耗的时间也不能减少, 因为 Spark 默认不会把 DS 放到内存之中.
来源: https://www.flyml.net/2018/02/07/spark-cache-example/