查找具有 Null 值的列并将它们写入 Pyspark 中每条记录的新列中

Question

我有一个场景，我必须为每条记录查找具有 Null 值的列，并将所有此类列名称写入单独的列中。

示例： 我有这个数据框：

+---------+---+------------+-----------+------+-------+
|firstName|age|jobStartDate|isGraduated|gender| salary|
+---------+---+------------+-----------+------+-------+
|     null|se3|  2006-01-01|          8|     M|      F|
|     null| a3|        null|       True|     F|   null|
|   Robert| 37|  1992-01-01|       null|     M|5000.50|
+---------+---+------------+-----------+------+-------+

预期结果应如下所示：

+---------+---+------------+-----------+------+-------+----------------------+
|firstName|age|jobStartDate|isGraduated|gender| salary|       Missing Columns|
+---------+---+------------+-----------+------+-------+----------------------+
|     null|se3|  2006-01-01|          8|     M|      F|             firstName|
|     null| a3|  2006-01-02|       True|     F|   null|      firstName,salary|
|   Robert| 37|  1992-01-01|       null|     M|5000.50|           isGraduated|
+---------+---+------------+-----------+------+-------+----------------------+

我编写了一半符合我预期结果的代码：

def find_exceptions(df,mand_cols = ['firstName','jobStartDate','salary']):
  miss = "Missing: "
  for column in mand_cols:
    if df[column] is None:
      miss = miss + column + ","
  return miss

我能够将缺失值收集为列表：

temp = sourceDF.rdd.map(find_exceptions)
temp.collect()
#result: 
['Missing: firstName,', 'Missing: firstName,jobStartDate,salary,', 'Missing: ']

我发现很难将其实际写入新专栏。我对 Spark 还很陌生，如果有人能帮我解决这个问题，我将不胜感激。

Answer 1

您可以分三步完成。

第 1 步：创建一个大小为列数的数组。如果条目为空，则将数组中的相应元素设置为列名的名称，否则将值保留为空。

第 2 步：过滤列名的数组

第 3 步：连接以具有逗号分隔的列表

df //step 1
    .withColumn("MissingColumns",
      array(
        when(col("firstName").isNull(),lit("firstName")),
        when(col("age").isNull(),lit("age")),
        when(col("jobStartDate").isNull(),lit("jobStartDate")),
        when(col("isGraduated").isNull(),lit("isGraduated")),
        when(col("gender").isNull(),lit("gender")),
        when(col("salary").isNull(),lit("salary"))
      )
    )
     //step 2
      .withColumn("MissingColumns",expr("filter(MissingColumns, c -> c IS NOT NULL)"))
     //step 3
      .withColumn("MissingColumns",concat_ws(",",col("MissingColumns")) )