Pyspark - 从每列中选择不同的值
Posted
技术标签:
【中文标题】Pyspark - 从每列中选择不同的值【英文标题】:Pyspark - Select the distinct values from each column 【发布时间】:2018-09-13 18:41:34 【问题描述】:我试图在数据框中的每一列中查找所有不同的值并显示在一个表中。
示例数据:
|-----------|-----------|-----------|
| COL_1 | COL_2 | COL_3 |
|-----------|-----------|-----------|
| A | C | D |
| A | C | D |
| A | C | E |
| B | C | E |
| B | C | F |
| B | C | F |
|-----------|-----------|-----------|
示例输出:
|-----------|-----------|-----------|
| COL_1 | COL_2 | COL_3 |
|-----------|-----------|-----------|
| A | C | D |
| B | | E |
| | | F |
|-----------|-----------|-----------|
这甚至可能吗?我已经能够在单独的表中执行此操作,但在一张表中会更好。
有什么想法吗?
【问题讨论】:
【参考方案1】:这里最简单的方法是在所有列上使用pyspark.sql.functions.collect_set
:
import pyspark.sql.functions as f
df.select(*[f.collect_set(c).alias(c) for c in df.columns]).show()
#+------+-----+---------+
#| COL_1|COL_2| COL_3|
#+------+-----+---------+
#|[B, A]| [C]|[F, E, D]|
#+------+-----+---------+
显然,这会将数据作为一行返回。
如果您希望得到您在问题中所写的输出(每列每个唯一值一行),这是可行的,但需要相当多的 pyspark 体操(并且任何解决方案都可能效率低得多)。
不过,我给你一些选择:
选项 1:分解并加入
您可以使用pyspark.sql.functions.posexplode
来分解每列值集中的元素以及数组中的索引。分别对每一列执行此操作,然后使用 functools.reduce
将生成的 DataFrame 列表外部连接在一起:
from functools import reduce
unique_row = df.select(*[f.collect_set(c).alias(c) for c in df.columns])
final_df = reduce(
lambda a, b: a.join(b, how="outer", on="pos"),
(unique_row.select(f.posexplode(c).alias("pos", c)) for c in unique_row.columns)
).drop("pos")
final_df.show()
#+-----+-----+-----+
#|COL_1|COL_2|COL_3|
#+-----+-----+-----+
#| A| null| E|
#| null| null| D|
#| B| C| F|
#+-----+-----+-----+
选项 2:按位置选择
首先计算最大数组的大小并将其存储在新列max_length
中。如果该索引处存在值,则从每个数组中选择元素。
我们再次使用pyspark.sql.functions.posexplode
,但这一次只是创建一个列来表示要提取的每个数组中的索引。
最后我们使用this trick,它允许您使用列值作为参数。
final_df= df.select(*[f.collect_set(c).alias(c) for c in df.columns])\
.withColumn("max_length", f.greatest(*[f.size(c) for c in df.columns]))\
.select("*", f.expr("posexplode(split(repeat(',', max_length-1), ','))"))\
.select(
*[
f.expr(
"case when size(c) > pos then c[pos] else null end AS c".format(c=c))
for c in df.columns
]
)
final_df.show()
#+-----+-----+-----+
#|COL_1|COL_2|COL_3|
#+-----+-----+-----+
#| B| C| F|
#| A| null| E|
#| null| null| D|
#+-----+-----+-----+
【讨论】:
以上是关于Pyspark - 从每列中选择不同的值的主要内容,如果未能解决你的问题,请参考以下文章