【发布时间】：2016-12-01 21:38:38
【问题描述】：
我想在 Spark 中运行代码时更好地理解 DAG 的执行。我知道 Spark 是惰性求值的，当我们执行任何操作（如计数、显示、缓存）时，它会运行转换命令。
但是我想知道在 DAG 中执行这些操作要追溯到多远。
就像我在预测数据帧上编写以下命令一样。
sorted_predictions=predictions.orderBy(['user','prediction'],ascending=[1,0])
def mapIdToString(x):
""" This function takes in the predicted dataframe and adds the original Item string to it
"""
global data_map_var
d_map=data_map_var.value
data_row= x.asDict()
#print data_row
for name, itemID in d_map.items():
if data_row['item']== itemID:
return (data_row['user'],data_row['item'],name,data_row['rating'],data_row['prediction'])
sorted_rdd=sorted_predictions.map(mapIdToString)
In [20]:
sorted_rdd.take(5)
Out[20]:
[(353, 21, u'DLR_Where Dreams Come True Town Hall', 0, 0.896152913570404),
(353, 2, u'DLR_Leading at a Higher Level', 1, 0.7186800241470337),
(353,
220,
u'DLR_The Year of a Million Dreams Leadership Update',
0,
0.687175452709198),
(353, 1, u'DLR_Challenging Conversations', 1, 0.6632049083709717),
(353,
0,
u'DLR_10 Keys to Inspiring, Engaging, and Energizing Your People',
1,
0.647541344165802)]
sorted_df=sqlContext.createDataFrame(sorted_rdd,['user','itemId','itemName','rating','prediction'])
sorted_df.registerTempTable("predictions_df")
query = """
select * from predictions_df
where user =353
and rating =0
"""
items_recommended=sqlContext.sql(query)
现在，当我运行以下命令时，我期待它，因为它是一个小查询，它应该运行得很快。但是要花很多时间才能输出。看起来它是一直走到 DAG 的顶部并再次执行所有的事情？
我不明白，因为当我执行 sorted_rdd.take(5) 命令时 DAG 会被破坏。因此，如果我现在运行以下命令，则该命令之后的任何内容都将被执行，而不是在此之前
items_recommended.count()
那为什么它运行了一个小时？我正在使用 60 个执行器和 5 个内核。 sorted_rdd 有 450MM 行。
编辑1：
这是对大卫回答的跟进。可以说我有以下命令。
对数据帧进行排序
sorted_predictions=predictions.orderBy(['user','prediction'],ascending=[1,0])
sorted_predictions.show(20)
sorted_rdd=sorted_predictions.map(mapIdToString)
sorted_rdd.take(5)
您是说每次我使用.take() 运行最后一个命令时，它都会返回到第一个orderBy 并再次对数据框进行排序并再次运行所有命令？即使我做了sorted_prediction.show() 来执行之前的排序命令？
编辑二：
如果我有如下功能：
def train_test_split(self,split_perc):
""" This function takes the DataFrame/RDD of ratings and splits
it into Training, Validation and testing based on the splitting
percentage passed as parameters
Param: ratings Dataframe of Row[(UserID,ItemID,ratings)]
Returns: train, validation, test
"""
# Converting the RDD back to dataframe to be used in DataFrame ml API
#ratings=sqlContext.createDataFrame(split_sdf,["user", "item", "rating"])
random_split=self.ratings_sdf.randomSplit(split_perc,seed=20)
#return random_split[0],random_split[1],random_split[2]
self.train=random_split[0]
self.train.cache().count()
# Converting the ratings column to float values for Validation and Test data
self.validation=random_split[1].withColumn('rating',(random_split[1].rating>0).astype('double'))
self.test=random_split[2].withColumn('rating',(random_split[2].rating>0).astype('double'))
self.validation.cache().count()
self.test.cache()
这个函数基本上是将一个数据帧分成训练、验证和测试。稍后我将在机器学习任务中使用所有这三个，因此将使用 train 来训练算法和 val 以进行超参数调整。
所以我缓存了以上三个。但是，为了使缓存可执行，我对所有三个都做了 .count。但是现在这个功能需要很长时间才能工作。你认为这三个都需要一个 .count 还是我可以只做一个 .count 一个 (test.count() 它会执行上述函数中的所有命令，并且也会缓存 train 和 val 数据报？我觉得应该可以并且不需要不必要的三个计数？
【问题讨论】：
-
试图解决后续问题
标签： python apache-spark pyspark