我是一个新手,我想在源数据帧(从JSON文件加载)下面进行转换:
+--+-----+-----+
|A |count|major|
+--+-----+-----+
| a| 1| m1|
| a| 1| m2|
| a| 2| m3|
| a| 3| m4|
| b| 4| m1|
| b| 1| m2|
| b| 2| m3|
| c| 3| m1|
| c| 4| m3|
| c| 5| m4|
| d| 6| m1|
| d| 1| m2|
| d| 2| m3|
| d| 3| m4|
| d| 4| m5|
| e| 4| m1|
| e| 5| m2|
| e| 1| m3|
| e| 1| m4|
| e| 1| m5|
+--+-----+-----+
放入以下结果数据框:
+--+--+--+--+--+--+
|A |m1|m2|m3|m4|m5|
+--+--+--+--+--+--+
| a| 1| 1| 2| 3| 0|
| b| 4| 2| 1| 0| 0|
| c| 3| 0| 4| 5| 0|
| d| 6| 1| 2| 3| 4|
| e| 4| 5| 1| 1| 1|
+--+--+--+--+--+--+
这是转换规则:
结果数据帧由以下A + (n major columns)
各major
列名称组成:
sorted(src_df.map(lambda x: x[2]).distinct().collect())
结果数据帧包含m
行,其中A
列的值由以下行提供:
sorted(src_df.map(lambda x: x[0]).distinct().collect())
在结果数据帧的每个主要列中的值是从源数据帧上的相应的值A
和主要(例如中第1行中源数据帧计数被映射到box
哪里A
是a
和列m1
)
A
和major
在源数据帧中的组合没有重复(请在SQL的两列中将其视为主键)
让我们从示例数据开始:
df = sqlContext.createDataFrame([
("a", 1, "m1"), ("a", 1, "m2"), ("a", 2, "m3"),
("a", 3, "m4"), ("b", 4, "m1"), ("b", 1, "m2"),
("b", 2, "m3"), ("c", 3, "m1"), ("c", 4, "m3"),
("c", 5, "m4"), ("d", 6, "m1"), ("d", 1, "m2"),
("d", 2, "m3"), ("d", 3, "m4"), ("d", 4, "m5"),
("e", 4, "m1"), ("e", 5, "m2"), ("e", 1, "m3"),
("e", 1, "m4"), ("e", 1, "m5")],
("a", "cnt", "major"))
请注意,我已更改count
为cnt
。Count是大多数SQL方言中的保留关键字,对于列名而言,它不是一个好的选择。
至少有两种方法可以重塑此数据:
在DataFrame上聚合
from pyspark.sql.functions import col, when, max
majors = sorted(df.select("major")
.distinct()
.map(lambda row: row[0])
.collect())
cols = [when(col("major") == m, col("cnt")).otherwise(None).alias(m)
for m in majors]
maxs = [max(col(m)).alias(m) for m in majors]
reshaped1 = (df
.select(col("a"), *cols)
.groupBy("a")
.agg(*maxs)
.na.fill(0))
reshaped1.show()
## +---+---+---+---+---+---+
## | a| m1| m2| m3| m4| m5|
## +---+---+---+---+---+---+
## | a| 1| 1| 2| 3| 0|
## | b| 4| 1| 2| 0| 0|
## | c| 3| 0| 4| 5| 0|
## | d| 6| 1| 2| 3| 4|
## | e| 4| 5| 1| 1| 1|
## +---+---+---+---+---+---+
groupBy
超过RDD
from pyspark.sql import Row
grouped = (df
.map(lambda row: (row.a, (row.major, row.cnt)))
.groupByKey())
def make_row(kv):
k, vs = kv
tmp = dict(list(vs) + [("a", k)])
return Row(**{k: tmp.get(k, 0) for k in ["a"] + majors})
reshaped2 = sqlContext.createDataFrame(grouped.map(make_row))
reshaped2.show()
## +---+---+---+---+---+---+
## | a| m1| m2| m3| m4| m5|
## +---+---+---+---+---+---+
## | a| 1| 1| 2| 3| 0|
## | e| 4| 5| 1| 1| 1|
## | c| 3| 0| 4| 5| 0|
## | b| 4| 1| 2| 0| 0|
## | d| 6| 1| 2| 3| 4|
## +---+---+---+---+---+---+
本文收集自互联网,转载请注明来源。
如有侵权,请联系[email protected] 删除。
我来说两句