PySpark：使用 RDD 并行化的输出与原始 RDD 一起执行其他操作答案

【问题标题】：PySpark: use RDD parallelization's output along with original RDD to perform other operationsPySpark：使用 RDD 并行化的输出与原始 RDD 一起执行其他操作
【发布时间】：2017-08-29 06:18:52
【问题描述】：

假设我有一个 RDD。在这个 RDD 上，我执行一些输出一些输出的操作。现在，我需要这个输出和原始 RDD 来执行一些其他操作。

怎么做？

这是我的代码：

rdd = sc.parallelize(input)
rdd1 = rdd.map(...)
...
output1 =  rdd1.collect() # output I need

output2 = rdd.map(some operations using output1)

【问题讨论】：

除非你真的需要，否则永远不要执行 collect()，因为这可能会使主集群的内存过载。除非您具体说明您要执行哪些操作，否则很难帮助您并行化此操作。
正确 - 原则上，您应该直接以rdd2 - rdd1.map(...) 继续，无需收集
@MohammedKashif 让我提供更多细节。假设我有一个看起来像 [['a', 1, [1, 2]], ['a', 2, [1, 1]], ['a', 3, [2, 2]], ['b', 4, [2, 2]]] 的 RDD（每个内部列表都组织为 ['key', 'id', 'vector']）。我想做两件事。首先我想为每个键（'a' 和 'b'）计算向量维度的最常见值，即我想输出类似center={'a': [1, 2], 'b': [2, 2]} 的东西。然后，我想计算每个向量到它的中心的距离，即dist([1, 2], center['a'])、dist([1, 1], center['a'])、...、dist([2, 2], center['b'])。
collect 将所有内容带到驱动程序的内存中。您应该使用substract 或join。大多数情况下，如果您从同一张表开始，这意味着您可能可以使用数据框和窗口函数进行计算。这将是最有效的，因为它产生一个 sort by key 而不是两个。

标签： python apache-spark pyspark rdd

【解决方案1】：

带有窗口函数：

在开始之前，让我们将 rdd 转换为数据框：

df = spark.createDataFrame(
    sc.parallelize(
        [['a', 1, [1, 2]], ['a', 2, [1, 1]], ['a', 3, [2, 2]], ['b', 4, [2, 2]]]
    ), ['c1', 'c2', 'c3']
)

首先我们计算出现次数：

from pyspark.sql import Window
import pyspark.sql.functions as psf
w1 = Window.partitionBy("c1", df.c3[0])
w2 = Window.partitionBy("c1", df.c3[1])
df1 = df.select(
    "c1", "c2", "c3", 
    psf.count("*").over(w1).alias("count1"), 
    psf.count("*").over(w2).alias("count2")
)

接下来我们找到最频繁的项目：

w1 = Window.partitionBy("c1").orderBy(psf.desc("count1"))
w2 = Window.partitionBy("c1").orderBy(psf.desc("count2"))
df2 = df1.select(
    "c1", "c2", "c3",
    psf.first(df1.c3[0]).over(w1).alias("most_freq1"), 
    psf.first(df1.c3[1]).over(w2).alias("most_freq2") 
)

然后，我们计算距离

df3 = df2.withColumn(
    "dist", 
    psf.sqrt((df2.most_freq1 - df2.c3[0])**2 + (df2.most_freq2 - df2.c3[1])**2)
)
df3.show()
    +---+---+------+----------+----------+----+
    | c1| c2|    c3|most_freq1|most_freq2|dist|
    +---+---+------+----------+----------+----+
    |  b|  4|[2, 2]|         2|         2| 0.0|
    |  a|  1|[1, 2]|         1|         2| 0.0|
    |  a|  3|[2, 2]|         1|         2| 1.0|
    |  a|  2|[1, 1]|         1|         2| 1.0|
    +---+---+------+----------+----------+----+

【讨论】：