PySpark或SQL：消耗合并

编程问答 2022-08-16

问题描述

我正在尝试将pyspark数据框或sql表中的多个输入列合并为多个输出列。

每个输出列将包含“第一个可用”输入值，然后“消费”它，因此该输入值不适用于随后的输出列。

+----+-----+-----+-----+-----+-----+---+------+------+------+
| ID | in1 | in2 | in3 | in4 | in5 | / | out1 | out2 | out3 |
+----+-----+-----+-----+-----+-----+---+------+------+------+
|  1 |     |     | C   |     |     | / | C    |      |      |
|  2 | A   |     | C   |     | E   | / | A    | C    | E    |
|  3 | A   | B   | C   |     |     | / | A    | B    | C    |
|  4 | A   | B   | C   | D   | E   | / | A    | B    | C    |
|  5 |     |     |     |     |     | / |      |      |      |
|  6 |     | B   |     |     | E   | / | B    | E    |      |
|  7 |     | B   |     | D   | E   | / | B    | D    | E    |
+----+-----+-----+-----+-----+-----+---+------+------+------+

执行此操作的最佳方法是什么？

编辑：澄清-in1，in2，in3等。可以是任何值

解决方法

这是方法。

import pyspark.sql.functions as f

df = spark.read.option("header","true").option("inferSchema","true").csv("test.csv")

cols = df.columns
cols.remove('ID')

df2 = df.withColumn('ins',f.array_except(f.array(*cols),f.array(f.lit(None))))

for i in range(0,3):
    df2 = df2.withColumn('out' + str(i+1),f.col('ins')[i])
    
df2.show(10,False)

+---+----+----+----+----+----+---------------+----+----+----+
|ID |in1 |in2 |in3 |in4 |in5 |ins            |out1|out2|out3|
+---+----+----+----+----+----+---------------+----+----+----+
|1  |null|null|C   |null|null|[C]            |C   |null|null|
|2  |A   |null|C   |null|E   |[A,C,E]      |A   |C   |E   |
|3  |A   |B   |C   |null|null|[A,B,C]      |A   |B   |C   |
|4  |A   |B   |C   |D   |E   |[A,D,E]|A   |B   |C   |
|5  |null|null|null|null|null|[]             |null|null|null|
|6  |null|B   |null|null|E   |[B,E]         |B   |E   |null|
|7  |null|B   |null|D   |E   |[B,E]      |B   |D   |E   |
+---+----+----+----+----+----+---------------+----+----+----+
,

以下答案假定输入列中的value包含二值布尔逻辑，这意味着in1只能是{'A'，None}，in2 {'B'，无}，等等。

import pyspark.sql.functions as F

df = spark.createDataFrame([[1,None,'C','D',None],[2,'A','E'],[3,'B',[4,[5,None]],['ID','in1','in2','in3','in4','in5'])

df2 = df.withColumn('out1',F.when(F.col('in1').isNotNull(),F.col('in1'))
                             .when(F.col('in2').isNotNull(),F.col('in2'))
                             .when(F.col('in3').isNotNull(),F.col('in3'))
                             .when(F.col('in4').isNotNull(),F.col('in4'))
                             .when(F.col('in5').isNotNull(),F.col('in5'))
                   )\
        .withColumn('out2',F.when(F.col('out1') < F.col('in2'),F.col('in2'))
                             .when(F.col('out1') < F.col('in3'),F.col('in3'))
                             .when(F.col('out1') < F.col('in4'),F.col('in4'))
                             .when(F.col('out1') < F.col('in5'),F.col('in5'))
                   )\
        .withColumn('out3',F.when(F.col('out2') < F.col('in3'),F.col('in3'))
                             .when(F.col('out2') < F.col('in4'),F.col('in4'))
                             .when(F.col('out2') < F.col('in5'),F.col('in5'))
                   )

df2.show()
>>>
+---+----+----+---+----+----+----+----+----+
| ID| in1| in2|in3| in4| in5|out1|out2|out3|
+---+----+----+---+----+----+----+----+----+
|  1|null|null|  C|   D|null|   C|   D|null|
|  2|   A|null|  C|null|   E|   A|   C|   E|
|  3|null|   B|  C|null|   E|   B|   C|   E|
|  4|null|null|  C|null|   E|   C|   E|null|
|  5|   A|   B|  C|null|null|   A|   B|   C|
+---+----+----+---+----+----+----+----+----+

上面的答案利用了这样一个事实：对于F.when(F.col('out1') < F.col('in2'),F.col('in2'))这样的表达式，只有out1和in2都不为空时，条件表达式才解析为True。 }仅在out2已填充的情况下填充。

pysparksqlsql