用 DataFrame 中的 None/null 值替换空字符串

Posted 2023-02-23

【中文标题】用 DataFrame 中的 None/null 值替换空字符串

【英文标题】：Replace empty strings with None/null values in DataFrame

【发布时间】：2016-01-22 03:17:06

【问题描述】：

我有一个Spark 1.5.0 DataFrame，在同一列中混合了null 和空字符串。我想将所有列中的所有空字符串转换为null（None，在 Python 中）。 DataFrame 可能有数百列，因此我试图避免对每一列进行硬编码操作。

请参阅下面的尝试，这会导致错误。

from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)

## Create a test DataFrame
testDF = sqlContext.createDataFrame([Row(col1='foo', col2=1), Row(col1='', col2=2), Row(col1=None, col2='')])
testDF.show()
## +----+----+
## |col1|col2|
## +----+----+
## | foo|   1|
## |    |   2|
## |null|null|
## +----+----+

## Try to replace an empty string with None/null
testDF.replace('', None).show()
## ValueError: value should be a float, int, long, string, list, or tuple

## A string value of null (obviously) doesn't work...
testDF.replace('', 'null').na.drop(subset='col1').show()
## +----+----+
## |col1|col2|
## +----+----+
## | foo|   1|
## |null|   2|
## +----+----+

【问题讨论】：

@palsch，不，它不返回列表。它返回一个数据帧。我用 Spark 文档的链接更新了问题。

@palsch 这不是一般的 Python 问题！ Spark DataFrames 是分布式数据结构，通常用于对大数据进行大量数据分析。所以你的解决方案不合适。

@eliasah 说实话，Pythonic lambda x: None if not x else x 用 udf 包裹就可以了 :)

@zero323 但他要求 OP 返回一个列表...

哪个答案最有效？

【参考方案1】：

就这么简单：

from pyspark.sql.functions import col, when

def blank_as_null(x):
    return when(col(x) != "", col(x)).otherwise(None)

dfWithEmptyReplaced = testDF.withColumn("col1", blank_as_null("col1"))

dfWithEmptyReplaced.show()
## +----+----+
## |col1|col2|
## +----+----+
## | foo|   1|
## |null|   2|
## |null|null|
## +----+----+

dfWithEmptyReplaced.na.drop().show()
## +----+----+
## |col1|col2|
## +----+----+
## | foo|   1|
## +----+----+

如果您想填充多列，例如可以减少：

to_convert = set([...]) # Some set of columns

reduce(lambda df, x: df.withColumn(x, blank_as_null(x)), to_convert, testDF)

或使用理解：

exprs = [
    blank_as_null(x).alias(x) if x in to_convert else x for x in testDF.columns]

testDF.select(*exprs)

如果要专门对字符串字段进行操作，请查看the answer by robin-loxley。

【讨论】：

谢谢@zero323。您的答案可以扩展到自动有效地处理许多列吗？也许列出所有列名，为每一列生成与您的答案相似的代码，然后评估代码？

我看不出有什么理由你不能。 DataFrames 被懒惰地评估，其余的只是一个标准的 Python。您会在编辑中找到一些选项。

我会接受这个答案，但您能否先添加来自@RobinLoxley 的内容？或者，如果您不介意，我可以编辑您的答案。

@dnlbrky 这不公平。

声明.otherwise(None) 不是必需的。对于不匹配的条件，始终返回 None（参见 spark.apache.org/docs/latest/api/python/…）

【参考方案2】：

UDF 的效率并不高。使用内置方法的正确方法是：

df = df.withColumn('myCol', when(col('myCol') == '', None).otherwise(col('myCol')))

【讨论】：

我收到一个“str”不可调用错误。任何想法为什么？

检查括号

嗯我直接从这里复制过来的。

我刚刚测试了代码，它是有效的。该错误可能是在数据帧的操作中的其他地方引入的，并且仅在诸如 collect() 或 show() 之类的“操作”之后才引发错误。如果不包含我的代码并运行 df.show()，是否会出现相同的错误？

这绝对是正确的解决方案，使用内置函数可以在 spark 端进行大量优化。 Python UDF 非常昂贵，因为 spark 执行器（无论您是否使用 pyspark，它始终在 JVM 上运行）需要序列化每一行（确切地说是成批的行），通过套接字将其发送到子 python 进程，评估您的 python 函数，序列化结果并从套接字中读取它

【参考方案3】：

只需添加 zero323 和 soulmachine 的答案。为所有 StringType 字段进行转换。

from pyspark.sql.types import StringType
string_fields = []
for i, f in enumerate(test_df.schema.fields):
    if isinstance(f.dataType, StringType):
        string_fields.append(f.name)

【讨论】：

enumerate 的用途是什么？我的意思是，我知道它的作用，但有没有理由使用它而不是 for field in test_df.schema.fields: ？

【参考方案4】：

我的解决方案比我目前看到的所有解决方案都要好得多，可以处理任意多的字段，看小函数如下：

  // Replace empty Strings with null values
  private def setEmptyToNull(df: DataFrame): DataFrame = 
    val exprs = df.schema.map  f =>
      f.dataType match 
        case StringType => when(length(col(f.name)) === 0, lit(null: String).cast(StringType)).otherwise(col(f.name)).as(f.name)
        case _ => col(f.name)
      
    

    df.select(exprs: _*)
  

你可以很容易地用 Python 重写上面的函数。

我从@liancheng学到了这个技巧

【参考方案5】：

如果您使用的是 python，您可以检查以下内容。

+----+-----+----+
|  id| name| age|
+----+-----+----+
|null|name1|  50|
|   2|     |    |
|    |name3|null|
+----+-----+----+

def convertToNull(dfa):
   for i in dfa.columns:
    dfa = dfa.withColumn(i , when(col(i) == '', None ).otherwise(col(i)))
  return dfa

convertToNull(dfa).show()

+----+-----+----+
|  id| name| age|
+----+-----+----+
|null|name1|  50|
|   2| null|null|
|null|name3|null|
+----+-----+----+

【参考方案6】：

我会在@zero323 的solution 中添加一个trim 来处理多个空格的情况：

def blank_as_null(x):
    return when(trim(col(x)) != "", col(x))

【参考方案7】：

这是 soulmachine 解决方案的不同版本，但我认为您不能轻易将其转换为 Python：

def emptyStringsToNone(df: DataFrame): DataFrame = 
  df.schema.foldLeft(df)(
    (current, field) =>
      field.dataType match 
        case DataTypes.StringType =>
          current.withColumn(
            field.name,
            when(length(col(field.name)) === 0, lit(null: String)).otherwise(col(field.name))
          )
        case _ => current
      
  )