一个Spark缓存的使用示例

发表于:2018-2-12 09:41  作者:佚名   来源:网络大数据

字体: | 上一篇 | 下一篇 |我要投稿 | 推荐标签: 软件开发 大数据

  之前一直不是非常理解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.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放到内存之中.

上文内容不用于商业目的,如涉及知识产权问题,请权利人联系博为峰小编(021-64471599-8017),我们将立即处理。

评 论

论坛新帖

顶部 底部


建议使用IE 6.0以上浏览器,800×600以上分辨率,法律顾问:上海瀛东律师事务所 张楠律师
版权所有 上海博为峰软件技术股份有限公司 Copyright©51testing.com 2003-2018, 沪ICP备05003035号
投诉及意见反馈:webmaster@51testing.com; 业务联系:service@51testing.com 021-64471599-8017

沪公网安备 31010102002173号

51Testing官方微信

51Testing官方微博

扫一扫 测试知识全知道