python - 了解 Spark 执行中的 DAG

Question

我想在 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 数据报？我觉得这应该可行并且不需要不必要的三计数？

Answer 1

Which I don't understand since the DAG would have been broken when I executed sorted_rdd.take(5) command. So anything after this command would be executed and not before that

Spark 的惰性计算扩展到将内容存储在内存中。除非您显式 cache() 中间数据，否则它不会这样做。如果没有 cache() 调用，Spark 也需要重新处理 take(5) 调用之前的所有步骤。要解决此问题，请在像这样的 take 操作之前缓存您的 rdd

 sorted_rdd.cache().take(5)

解决编辑问题

Are you saying every time I run the last command with .take() it will go back to the first orderBy and sort the data frame again and run all the commands again?. Even if I did sorted_prediction.show() to execute the earlier sorting command?.

正确。在下面的代码中，Spark 需要运行创建预测的所有步骤以及orderBy计算以显示 20 行sorted_predictions。然后，它将运行创建预测、orderBy计算和map计算的所有步骤，以显示 5 行sorted_rdd。

sorted_predictions=predictions.orderBy(['user','prediction'],ascending=[1,0])
sorted_predictions.show(20)

sorted_rdd=sorted_predictions.map(mapIdToString)
sorted_rdd.take(5)

来自评论

I thought cache is also an action

缓存本身并不是一个操作。这是一条将 RDD/DataFrame 存储在内存中的指令，但这实际上不会发生，直到运行某个操作(例如 count、take、show 等)