如何在 Pyspark 中不使用 collect() 方法将 pyspark.rdd.PipelinedRDD 转换为数据框？答案

【问题标题】：How to convert pyspark.rdd.PipelinedRDD to Data frame with out using collect() method in Pyspark?如何在 Pyspark 中不使用 collect() 方法将 pyspark.rdd.PipelinedRDD 转换为数据框？
【发布时间】：2018-06-15 02:47:24
【问题描述】：

我有pyspark.rdd.PipelinedRDD(Rdd1)。当我在做Rdd1.collect() 时，它给出的结果如下所示。

 [(10, {3: 3.616726727464709, 4: 2.9996439803387602, 5: 1.6767412921625855}),
 (1, {3: 2.016527311459324, 4: -1.5271512313750577, 5: 1.9665475696370045}),
 (2, {3: 6.230272144805092, 4: 4.033642544526678, 5: 3.1517805604906313}),
 (3, {3: -0.3924680103722977, 4: 2.9757316477407443, 5: -1.5689126834176417})]

现在我想在不使用 collect() 方法的情况下将 pyspark.rdd.PipelinedRDD 转换为 Data frame

我的最终数据框应该如下所示。 df.show() 应该是这样的：

+----------+-------+-------------------+
|CId       |IID    |Score              |
+----------+-------+-------------------+
|10        |4      |2.9996439803387602 |
|10        |5      |1.6767412921625855 |
|10        |3      |3.616726727464709  |
|1         |4      |-1.5271512313750577|
|1         |5      |1.9665475696370045 |
|1         |3      |2.016527311459324  |
|2         |4      |4.033642544526678  |
|2         |5      |3.1517805604906313 |
|2         |3      |6.230272144805092  |
|3         |4      |2.9757316477407443 |
|3         |5      |-1.5689126834176417|
|3         |3      |-0.3924680103722977|
+----------+-------+-------------------+

接下来应用收集、迭代和最后的数据帧，我可以实现这种转换为 rdd。

但现在我想在不使用任何collect() 方法的情况下将pyspark.rdd.PipelinedRDD 转换为Dataframe。

请告诉我如何实现这一目标？

【问题讨论】：

标签： python-3.x apache-spark pyspark apache-spark-sql rdd

【解决方案1】：

你想在这里做两件事： 1. 扁平化你的数据 2. 放入数据框

一种方法如下：

首先，让我们展平字典：

rdd2 = Rdd1.flatMapValues(lambda x : [ (k, x[k]) for k in x.keys()])

在收集数据时，您会得到如下信息：

[(10, (3, 3.616726727464709)), (10, (4, 2.9996439803387602)), ...

然后我们可以将数据格式化，转成dataframe：

rdd2.map(lambda x : (x[0], x[1][0], x[1][1]))\
    .toDF(("CId", "IID", "Score"))\
    .show()

给你这个：

+---+---+-------------------+
|CId|IID|              Score|
+---+---+-------------------+
| 10|  3|  3.616726727464709|
| 10|  4| 2.9996439803387602|
| 10|  5| 1.6767412921625855|
|  1|  3|  2.016527311459324|
|  1|  4|-1.5271512313750577|
|  1|  5| 1.9665475696370045|
|  2|  3|  6.230272144805092|
|  2|  4|  4.033642544526678|
|  2|  5| 3.1517805604906313|
|  3|  3|-0.3924680103722977|
|  3|  4| 2.9757316477407443|
|  3|  5|-1.5689126834176417|
+---+---+-------------------+

【讨论】：

【解决方案2】：

有一个更简单、更优雅的解决方案可以避免使用 python lambda 表达式，如 @oli 答案，它依赖于 spark DataFrames 的 explode，它完全符合您的要求。它也应该更快，因为不需要使用 python lambda 两次。见下文：

from pyspark.sql.functions import explode

# dummy data
data = [(10, {3: 3.616726727464709, 4: 2.9996439803387602, 5: 1.6767412921625855}),
        (1, {3: 2.016527311459324, 4: -1.5271512313750577, 5: 1.9665475696370045}),
        (2, {3: 6.230272144805092, 4: 4.033642544526678, 5: 3.1517805604906313}),
        (3, {3: -0.3924680103722977, 4: 2.9757316477407443, 5: -1.5689126834176417})]

# create your rdd
rdd = sc.parallelize(data)

# convert to spark data frame
df = rdd.toDF(["CId", "Values"])

# use explode
df.select("CId", explode("Values").alias("IID", "Score")).show()

+---+---+-------------------+
|CId|IID|              Score|
+---+---+-------------------+
| 10|  3|  3.616726727464709|
| 10|  4| 2.9996439803387602|
| 10|  5| 1.6767412921625855|
|  1|  3|  2.016527311459324|
|  1|  4|-1.5271512313750577|
|  1|  5| 1.9665475696370045|
|  2|  3|  6.230272144805092|
|  2|  4|  4.033642544526678|
|  2|  5| 3.1517805604906313|
|  3|  3|-0.3924680103722977|
|  3|  4| 2.9757316477407443|
|  3|  5|-1.5689126834176417|
+---+---+-------------------+

【讨论】：

【解决方案3】：

这就是使用 scala 的方法

  val Rdd1 = spark.sparkContext.parallelize(Seq(
    (10, Map(3 -> 3.616726727464709, 4 -> 2.9996439803387602, 5 -> 1.6767412921625855)),
    (1, Map(3 -> 2.016527311459324, 4 -> -1.5271512313750577, 5 -> 1.9665475696370045)),
    (2, Map(3 -> 6.230272144805092, 4 -> 4.033642544526678, 5 -> 3.1517805604906313)),
    (3, Map(3 -> -0.3924680103722977, 4 -> 2.9757316477407443, 5 -> -1.5689126834176417))
  ))

  val x = Rdd1.flatMap(x => (x._2.map(y => (x._1, y._1, y._2))))
         .toDF("CId", "IId", "score")

输出：

+---+---+-------------------+
|CId|IId|score              |
+---+---+-------------------+
|10 |3  |3.616726727464709  |
|10 |4  |2.9996439803387602 |
|10 |5  |1.6767412921625855 |
|1  |3  |2.016527311459324  |
|1  |4  |-1.5271512313750577|
|1  |5  |1.9665475696370045 |
|2  |3  |6.230272144805092  |
|2  |4  |4.033642544526678  |
|2  |5  |3.1517805604906313 |
|3  |3  |-0.3924680103722977|
|3  |4  |2.9757316477407443 |
|3  |5  |-1.5689126834176417|
+---+---+-------------------+

希望你能转换成 pyspark。

【讨论】：

【解决方案4】：

确保首先创建火花会话：

sc = SparkContext()
spark = SparkSession(sc)

我在尝试解决这个确切问题时找到了这个答案。
'PipelinedRDD' object has no attribute 'toDF' in PySpark

【讨论】：