如何替换 PySpark DataFrame 中的无穷大答案

【问题标题】：How to replace infinity in PySpark DataFrame如何替换 PySpark DataFrame 中的无穷大
【发布时间】：2016-03-29 17:31:38
【问题描述】：

似乎不支持替换无穷大值。我尝试了下面的代码，但它不起作用。还是我错过了什么？

a=sqlContext.createDataFrame([(None, None), (1, np.inf), (None, 2)])
a.replace(np.inf, 10)

还是要走痛苦的路：将PySpark DataFrame转成pandas DataFrame，替换无穷大值，再转回PySpark DataFrame

【问题讨论】：

标签： python pandas apache-spark pyspark apache-spark-sql

【解决方案1】：

似乎不支持替换无穷大值。

实际上它看起来像一个 Py4J 错误，而不是 replace 本身的问题。见Support nan/inf between Python and Java。

作为一种解决方法，您可以尝试任一 UDF（慢速选项）：

from pyspark.sql.types import DoubleType
from pyspark.sql.functions import col, lit, udf, when

df = sc.parallelize([(None, None), (1.0, np.inf), (None, 2.0)]).toDF(["x", "y"])

replace_infs_udf = udf(
    lambda x, v: float(v) if x and np.isinf(x) else x, DoubleType()
)

df.withColumn("x1", replace_infs_udf(col("y"), lit(-99.0))).show()

## +----+--------+-----+
## |   x|       y|   x1|
## +----+--------+-----+
## |null|    null| null|
## | 1.0|Infinity|-99.0|
## |null|     2.0|  2.0|
## +----+--------+-----+

或者这样的表达方式：

def replace_infs(c, v):
    is_infinite = c.isin([
        lit("+Infinity").cast("double"),
        lit("-Infinity").cast("double")
    ])
    return when(c.isNotNull() & is_infinite, v).otherwise(c)

df.withColumn("x1", replace_infs(col("y"), lit(-99))).show()

## +----+--------+-----+
## |   x|       y|   x1|
## +----+--------+-----+
## |null|    null| null|
## | 1.0|Infinity|-99.0|
## |null|     2.0|  2.0|
## +----+--------+-----+

【讨论】：

为什么UDFs 比表达式慢？
@AlbertoBonsanto 因为DataFrame 不是 Python 对象，所以它需要一个完整的往返行程。
@AlbertoBonsanto 另一个与 PySpark 无关的方面是 UDF 只是优化器的一个黑匣子。在这里应该没关系，但总的来说，这意味着您无法推断需要 UDF 的操作。最后，据我所知，内部表示不使用标准的 Scala 类型。因此，即使在 Scala 或 Java 中，您可能更喜欢直接使用不带 UDF 的表达式。
谢谢，问题是有时我在弄清楚如何使用表达式而不是 UDF 来实现类似的事情时遇到问题，我之所以问是因为我有一个将字母数组转换为 @ 的代码987654328@ 使用UDFs 并且代码从未完成
@KevinGhaboosi 这是由于类型不匹配。 Spark 不将 Python 整数视为 double / float 列的有效值。已修复（感谢您的编辑！）。