Pyspark：比较RDD的元素答案

【问题标题】：Pyspark: comparing elements of RDDPyspark：比较RDD的元素
【发布时间】：2021-10-07 21:32:47
【问题描述】：

类似于我在here 发布的关于使用 DF 的问题，我如何检索每个序列中的第一个元素，但在这种情况下使用 RDD？我想将每个项目与前一个项目进行比较。在序列后面重复的项目是可以接受的，即(67,375, 14:20:14) 可能会出现在 RDD 的后面，应该保留。

输入


(67, 312, 12:09:00)
(67, 375, 12:23:00)  
(67, 375, 12:25:00)    
(67, 650, 12:26:00)
(75, 650, 12:27:00)  
(75, 650, 12:29:00)    
(75, 800, 12:30:00)    
(67, 375, 14:20:14)

输出



(67, 312, 12:09:00)
(67, 375, 12:23:00)  
(67, 650, 12:26:00)
(75, 650, 12:27:00)  
(75, 800, 12:30:00)  
(67, 375, 14:20:14)

【问题讨论】：

标签： pyspark comparison rdd

【解决方案1】：

这会奏效。但是，我唯一担心的是，您不能依赖 rdd 上的转换将导致的输出顺序。因此，为了保留顺序，我强烈建议您按列排序，幸运的是，这里有时间戳列。

如果您不打算按时间戳排序，请使用数据框窗口方法。即使在那里，您也可能需要排序:)

rdd = sc.parallelize([(67, 312, "12:09:00"),
(67, 375, "12:23:00"), 
(67, 375, "12:25:00"),   
(67, 650, "12:26:00"),
(75, 650, "12:27:00"), 
(75, 650, "12:29:00"),   
(75, 800, "12:30:00") ])

# Fix 1st two columns as keys.
rdd_fix_keys = rdd.map(lambda x:((x[0],x[1]),(x[2])))

# Group the values of similar keys.
rdd_group_by_key = rdd_regroup_keys.reduceByKey(lambda x,y:(x,y))

# Pick first occurence of the grouped values, as per your requirement.
rdd_pick_first_occurence = rdd_group_by_key.map(lambda x:(x[0], x[1][0]) if not isinstance(x[1], str) else x)

# Sort by timestamp.
rdd_pick_first_occurence.map(lambda x:(x[0][0],x[0][1],x[1])).sortBy(lambda x: x[2]).collect()

注意：这里的顺序改了。

【讨论】：

谢谢，这肯定很接近，但我应该在示例中包含我需要保留的重复值。如果按时间排序，还应保留第二次出现的 67, 375 -> (67,375, 14:20:14)。我已经编辑了示例
@Tony Russa，我想知道，是什么阻止您将此 RDD 转换为数据框并进行这些转换？如果转换没问题，请告诉我，我们可以用 DF 轻松解决。
嗨，吉姆。是的，我可以轻松地将其转换为数据框并这样做。我试图通过 RDDs 解决它的原因是我是 pyspark 的新手，我正在使用 RDDs 和 Dataframes 解决一些问题，以尝试了解每个更适合的差异和用例，或者是否应该在每种情况下使用 DF。看起来这绝对是 DF 更适合的用例之一
它在 RDD 中仍然可行，编写复杂的用户定义函数。但是，当另一个舒适的 API（如 dataframe）可以满足您的要求时，为什么还要让自己紧张呢。恕我直言，当您的数据是结构化的并带有模式时，最好使用 DF 而不是 RDD。提示：您可以在 Dataframes 中使用 windowing、lead、lag 函数来实现这一点。万事如意！
如果答案解决了您最初的问题或给出了这样做的想法，请投票赞成。