如何将 pyspark rdd 转换为 Dataframe答案

【问题标题】：how to convert pyspark rdd into a Dataframe如何将 pyspark rdd 转换为 Dataframe
【发布时间】：2020-07-13 17:50:21
【问题描述】：

我有一个如下所示的数据框 df

df=

+---+---+----+---+---+
|  a|  b|   c|  d|  e|
+---+---+----+---+---+
|  1|  a|foo1|  4|  5|
|  2|  b| bar|  4|  6|
|  3|  c| mnc|  4|  7|
|  4|  c| mnc|  4|  7|
+---+---+----+---+---+

我想实现类似的目标 df1=

+---+---+-----------------------------------------------+
|  a|  b|   c                                           |
+---+---+-----------------------------------------------+
|  1|  a|{'a': 1, 'b': 'a', 'c': 'foo1', 'd': 4, 'e': 5}|                            
|  2|  b|{'a': 2, 'b': 'b', 'c': 'bar', 'd': 4, 'e': 6} |                                       
|  3|  c|{'a': 3, 'b': 'c', 'c': 'mnc', 'd': 4, 'e': 7} |                                       
|  4|  c|{'a': 4, 'b': 'c', 'c': 'mnc', 'd': 4, 'e': 7} |                                       
+---+---+-----------------------------------------------+

我真的很想避免 group by，所以我想首先将数据帧转换为 rdd，然后再将它们转换为一个数据帧

我写的代码是

df2=df.rdd.flatMap(lambda x:(x.a,x.b,x.asDict()))

在 df2 上执行 foreach 时，我得到的是 rdd 格式的结果所以我尝试用它创建一个数据框。

df3=df2.toDF() #1st way
df3=sparkSession.createDataframe(df2) #2nd way

但是我在这两种方式中都遇到了错误。有人可以解释一下我在这里做错了什么以及如何实现我的重新定位

【问题讨论】：

你能把你遇到的错误粘贴在这里吗？
raise TypeError("Can not infer schema for type: %s" % type(row)) TypeError: Can not infer schema for type:

标签： python dataframe apache-spark pyspark

【解决方案1】：

可以使用 spark sql 完成如下：

Spark SQL

data.createOrReplaceTempView("data")
spark.sql("""
select a, b, to_json(named_struct('a',a, 'b',b,'c',c,'d',d,'e',e)) as c
from data""").show(20,False)

输出

# +---+---+----------------------------------------+
# |a  |b  |c                                       |
# +---+---+----------------------------------------+
# |1  |a  |{"a":1,"b":"a","c":"foo1","d":"4","e":5}|
# |2  |b  |{"a":2,"b":"b","c":"bar","d":"4","e":6} |
# |3  |c  |{"a":3,"b":"c","c":"mnc","d":"4","e":7} |
# |4  |c  |{"a":4,"b":"c","c":"mnc","d":"4","e":7} |
# +---+---+----------------------------------------+

数据帧 API

result = data\
 .withColumn('c',to_json(struct(data.a,data.b,data.c,data.d,data.e)))\
 .select("a","b","c")
result.show(20,False)

输出

# +---+---+----------------------------------------+
# |a  |b  |c                                       |
# +---+---+----------------------------------------+
# |1  |a  |{"a":1,"b":"a","c":"foo1","d":"4","e":5}|
# |2  |b  |{"a":2,"b":"b","c":"bar","d":"4","e":6} |
# |3  |c  |{"a":3,"b":"c","c":"mnc","d":"4","e":7} |
# |4  |c  |{"a":4,"b":"c","c":"mnc","d":"4","e":7} |
# +---+---+----------------------------------------+

【讨论】：

我有很多列，所以我只是修改了 .withColumn('c', to_json(struct([data[x] for x in data.columns]))) 。这个答案很简单.谢谢
好的。有道理

【解决方案2】：

您可以从地图类型列创建 json 列

import pyspark.sql.functions as F
df = sqlContext.createDataFrame(
    [(0, 1, 23, 4, 8, 9, 5, "b1"), (1, 2, 43, 8, 10, 20, 43, "e1")], 
    ("id", "a1", "b1", "c1", "d1", "e1", "f1", "ref")
)
tst = [[F.lit(c),F.col(c)] for c in df.columns]
tst_flat =[item for sublist in tst for item in sublist]
#%%

map_coln = F.create_map(*tst_flat)

df1=df.withColumn("out",F.to_json(map_coln))

结果：

In [37]: df1.show(truncate=False)
+---+---+---+---+---+---+---+---+-------------------------------------------------------------------------------+
|id |a1 |b1 |c1 |d1 |e1 |f1 |ref|out                                                                            |
+---+---+---+---+---+---+---+---+-------------------------------------------------------------------------------+
|0  |1  |23 |4  |8  |9  |5  |b1 |{"id":"0","a1":"1","b1":"23","c1":"4","d1":"8","e1":"9","f1":"5","ref":"b1"}   |
|1  |2  |43 |8  |10 |20 |43 |e1 |{"id":"1","a1":"2","b1":"43","c1":"8","d1":"10","e1":"20","f1":"43","ref":"e1"}|
+---+---+---+---+---+---+---+---+-------------------------------------------------------------------------------+

【讨论】：