Pandas 数据帧到 Spark 数据帧，将 NaN 转换为实际空值？

Question

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

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

Answer 1

TL;DR目前最好的选择是完全跳过 Pandas。

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

另一只手的 Pandas 没有可用于表示缺失值的原生值。因此，它使用了诸如NaN / NaT 或Inf 之类的占位符，Spark 无法将它们与实际的NaNs 和Infs 区分开来，并且转换规则取决于列类型。唯一的例外是 object 列（通常是字符串），它可以包含 None 值。您可以从the documentation 了解有关处理 Pandas 缺失值的更多信息。

例如，pandas 中的 NaN 在转换为 Spark 数据帧时最终会变成字符串“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|
+----+----+-------------------+

Answer 2

如果您想加载 pandas df，您可以将 NaN 替换为 None：

import pandas as pd
def load_csv(spark, path):
    """read csv to spark df"""
    pd_df = pd.read_csv(path)
    pd_df = pd_df.where((pd.notnull(pd_df)), None)
    df = spark.createDataFrame(pd_df)
    return df