一个Spark缓存的使用示例
关于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.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放到内存之中.
时间:2018-10-09 22:35 来源: 转发量:次
声明：本站部分作品是由网友自主投稿和发布、编辑整理上传，对此类作品本站仅提供交流平台，转载的目的在于传递更多信息及用于网络分享，并不代表本站赞同其观点和对其真实性负责，不为其版权负责。如果您发现网站上有侵犯您的知识产权的作品，请与我们取得联系，我们会及时修改或删除。
相关文章：
相关推荐：
网友评论：
