Spark 运行 在非常小的数据集上非常慢
Spark running very slow on a very small data set
下面这个简单的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 助您一臂之力
以后强烈建议大家使用DataFrameapi，Spark会为你做一些优化，比如下推谓词优化。
最后但并非最不重要的一点是，使用自定义函数的成本也很高。如果您正在使用 DataFrames，请尝试仅使用 org.apache.spark.sql.functions 模块中的现有函数。
使用 Spark 分析代码 UI
我还建议通过 Spark UI 分析您的代码，因为这可能不是您的代码的问题，因为您的数据很小，而是节点的问题。
下面这个简单的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 助您一臂之力
以后强烈建议大家使用DataFrameapi，Spark会为你做一些优化，比如下推谓词优化。
最后但并非最不重要的一点是，使用自定义函数的成本也很高。如果您正在使用 DataFrames，请尝试仅使用 org.apache.spark.sql.functions 模块中的现有函数。
使用 Spark 分析代码 UI
我还建议通过 Spark UI 分析您的代码，因为这可能不是您的代码的问题，因为您的数据很小，而是节点的问题。