查找具有空值的列，并将它们写入Pyspark中每个记录的新列中

Question

我有一个场景，在这个场景中，我必须找到每个记录都有空值的列，并将所有这样的列名写入一个单独的列中

例如： 我有这个数据框：

+---------+---+------------+-----------+------+-------+
|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：创建一个大小为多少列的数组。如果条目为null，则将数组中的相应元素设置为列名的名称，否则将值保留为null

步骤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")) )