python - Spark 在非常小的数据集上运行非常慢
问题描述
以下简单的 spark 程序需要 4 分钟才能运行。我不知道这段代码有什么问题。
首先,我生成了一个非常小的 rdd
D = spark.sparkContext.parallelize([(0,[1,2,3]),(1,[2,3]),(2,[0,3]),(3,[1])]).cache()
然后我生成一个向量
P1 = spark.sparkContext.parallelize(list(zip(list(range(4)),[1/4]*4))).cache()
然后我定义了一个函数来执行该map
步骤
def MyFun(x):
L0 = len(x[2])
L = []
for i in x[2]:
L.append((i,x[1]/L0))
return L
然后我执行以下代码
P0 = P1
D0 = D.join(P1).map(lambda x: [x[0],x[1][1],x[1][0]]).cache()
C0 = D0.flatMap(lambda x: MyFun(x)).cache()
P1 = C0.reduceByKey(lambda x,y:x+y).mapValues(lambda x:x*1.2+3.4).sortByKey().cache()
Diff = P1.join(P0).map(lambda x: abs(x[1][0]-x[1][1])).sum()
鉴于我的数据是如此之小,我无法弄清楚这段代码运行如此缓慢的原因......
解决方案
我有一些建议可以帮助您加快这项工作。
仅在需要时缓存
缓存的过程就是把你创建的dag写到磁盘上。所以缓存每一步可能会花费很多,而不是加快进程。
我建议你cache
只做P1。
使用 DataFrames 让 Spark 帮助您
之后,我强烈建议你使用DataFrame
api,Spark 可以为你做一些优化,比如下推谓词优化。
最后但并非最不重要的一点是,使用自定义函数的成本也很高。如果您正在使用DataFrames
,请尝试仅使用org.apache.spark.sql.functions
模块中的现有功能。
使用 Spark UI 分析代码
我还建议通过 Spark UI 分析您的代码,因为它可能不是您的代码的问题,因为您的数据很少,而是节点的问题。
推荐阅读
- javascript - 如何使用 IE11 和 certenroll 向 PKCS10 证书请求添加 challengePassword?
- javascript - 检测图像上/外部的点击不起作用
- c++ - hello world 示例,用于在 C 中使用 SSL 的猫鼬网络服务器
- aerospike - 如何在 aerospike nodejs 客户端中运行多个谓词映射键过滤器?
- django - 在 Django 中动态从 view.py 中的引导手风琴菜单获取数据
- typescript - 删除函数不带参数值
- python - python'switches'(dict)是否应该调用内部的每个函数?
- python - 如何使用 Python 通过 knox 与 Hbase 交互?
- r - 带有“lm”的线性(对数对数)模型:如何获得预测值总和的预测方差
- azure - az vm run-command failed