我只需要附加那些在pyspark数据帧中具有非空值的人

Question

我有一个 pyspark 数据帧 （df） 具有以下示例表 （表 1）： id， col1， col2， col3 1， abc， 空， 定义 2， 空， 定义， 定义， abc 3， 定义， abc， 空

我试图通过忽略空值附加所有列来获取新列（final）。我已经尝试了pyspark代码并使用了f.array（col1、col2、col3）。值会被附加，但不会忽略空值。我还尝试了UDF仅附加非空列，但它不起作用。

import pyspark.sql.functions as f    
df = spark.table('table1')
df = df.withColumn('final', f.array(col1,col2,col3))

Actual result:
id, col1, col2, col3, final
1, abc, null, def, [abc,,def]
2, null, def, abc, [,def, abc]
3, def, abc, null, [def,abc,,]

expected result:
id, col1, col2, col3, final
1, abc, null, def, [abc,def]
2, null, def, abc, [def, abc]
3, def, abc, null, [def,abc]


my col1, col2, col3 schema are as below:
where as col1 name is applications


applications: struct (nullable = false)
    applicationid: string (nullable = true)
    createdat: string (nullable = true)
    updatedat: string (nullable = true)
    source_name: string (nullable = true)
    status: string (nullable = true)
    creditbureautypeid: string (nullable = true)
    score: integer (nullable = true)
    applicationcreditreportid: string (nullable = true)
    firstname: string (nullable = false)
    lastname: string (nullable = false)
    dateofbirth: string (nullable = false)
    accounts: array (nullable = true)
        element: struct (containsNull = true)
        applicationcreditreportaccountid: string (nullable = true)
        currentbalance: integer (nullable = true)
        institutionid: string (nullable = true)
        accounttypeid: string (nullable = true)
        dayspastdue: integer (nullable = true)
        institution_name: string (nullable = true)
        account_type_name: string (nullable = true) 

如果问题不清楚或需要更多信息，请告诉我。任何帮助都将不胜感激。:)

Answer 1

您可以定义自己的UDF如下：

def only_not_null(st,nd,rd):
   return [x for x in  locals().values() if x is not None]  # Take non empty columns

然后调用：

df = spark.table('table1')
df = df.withColumn('final', f.udf(only_not_null)(col1,col2,col3))

Answer 2

使用UDF

from pyspark.sql.functions import udf, array

def join_columns(row_list):
    return [cell_val for cell_val in row_list if cell_val is not None]

join_udf = udf(join_columns)

df = spark.table('table1')
df = df.withColumn('final', join_udf(array(col1,col2,col3))

适用于多个列，而不仅仅是3，只需编辑数组内的列。

Answer 3

从Spark 2.4开始，你可以使用高阶函数来实现(不需要UDF)。在PySpark中，查询可能是这样的:

result = (
    df
    .withColumn("temp", f.array("col1", "col2", "col3"))
    .withColumn("final", f.expr("FILTER(temp, x -> x is not null)"))
    .drop("temp")
)