python - Pandas 数据框到 Spark 数据框，处理 NaN 转换为实际空值？

Question

我想将数据帧从 pandas 转换为 spark，我正在使用 spark_context.createDataFrame() 方法来创建数据帧。我还在 createDataFrame() 方法中指定了架构。

我想知道的是如何处理特殊情况。例如，当转换为 Spark 数据帧时，pandas 中的 NaN 最终成为字符串“NaN”。我正在寻找如何获取实际空值而不是“NaN”的方法。

Answer 1

TL;DR您现在最好的选择是完全跳过 Pandas。

问题的根源在于 Pandas 的表现力不如 Spark SQL。 Spark 提供了 NULL(在 SQL 意义上，作为缺失值)和 NaN(数字不是数字)。

另一方面， Pandas 没有可用于表示缺失值的原生值。因此，它使用占位符，如 NaN/NaT 或 Inf，Spark 无法区分实际的 NaN 和Infs 和转换规则取决于列类型。唯一的异常(exception)是 object 列(通常是字符串)，它可以包含 None 值。您可以从 the documentation 了解更多关于处理缺失值的 Pandas。 .

For example, NaN in pandas when converted to Spark dataframe ends up being string "NaN".

这实际上是不正确的。取决于输入列的类型。如果列显示 NaN，则它很可能不是数字值，也不是纯字符串:

from pyspark.sql.functions import isnan, isnull

pdf = pd.DataFrame({
    "x": [1, None], "y": [None, "foo"], 
    "z": [pd.Timestamp("20120101"), pd.Timestamp("NaT")]
})
sdf = spark.createDataFrame(pdf)

sdf.show()

+---+----+-------------------+
|  x|   y|                  z|
+---+----+-------------------+
|1.0|null|2012-01-01 00:00:00|
|NaN| foo|               null|
+---+----+-------------------+

sdf.select([
    f(c) for c in sdf.columns for f in [isnan, isnull] 
    if (f, c) != (isnan, "z")  # isnan cannot be applied to timestamp 
]).show()

+--------+-----------+--------+-----------+-----------+
|isnan(x)|(x IS NULL)|isnan(y)|(y IS NULL)|(z IS NULL)|
+--------+-----------+--------+-----------+-----------+
|   false|      false|   false|       true|      false|
|    true|      false|   false|      false|       true|
+--------+-----------+--------+-----------+-----------+

在实践中，并行化的本地集合(包括 Pandas 对象)在简单测试和玩具示例之外的重要性可以忽略不计，因此您始终可以手动转换数据(跳过可能的 Arrow 优化):

import numpy as np

spark.createDataFrame([
   tuple(
        None if isinstance(x, (float, int)) and np.isnan(x) else x
        for x in record.tolist())
   for record in pdf.to_records(index=False)
], pdf.columns.tolist()).show()

+----+----+-------------------+
|   x|   y|                  z|
+----+----+-------------------+
| 1.0|null|1325376000000000000|
|null| foo|               null|
+----+----+-------------------+

如果丢失/非数字歧义不是问题，那么只需像往常一样加载数据并在 Spark 中替换。

from pyspark.sql.functions import col, when 

sdf.select([
    when(~isnan(c), col(c)).alias(c) if t in ("double", "float") else c 
    for c, t in sdf.dtypes
]).show()

+----+----+-------------------+
|   x|   y|                  z|
+----+----+-------------------+
| 1.0|null|2012-01-01 00:00:00|
|null| foo|               null|
+----+----+-------------------+