如何在 Spark Scala 中将 null NAN 或 Infinite 值替换为默认值

Question

我正在将 csvs 读入 Spark，并将架构设置为所有 DecimalType(10,0) 列。当我查询数据时，我收到以下错误：

NumberFormatException: Infinite or NaN

如果我的数据框中有 NaN/null/infinite 值，我想将它们设置为 0。我该怎么做？这就是我尝试加载数据的方式：

var cases = spark.read.option("header",false).
option("nanValue","0").
option("nullValue","0").
option("positiveInf","0").
option("negativeInf","0").
schema(schema).
csv(...

任何帮助将不胜感激。

Answer 1

如果多列有NaN值，可以使用na.fill()填充默认值

示例：

  val spark =
    SparkSession.builder().master("local").appName("test").getOrCreate()

  import spark.implicits._

  val data = spark.sparkContext.parallelize(
    Seq((0f,0f, "2016-01-1"),
        (1f,1f, "2016-02-2"),
        (2f,2f, "2016-03-21"),
        (Float.NaN,Float.NaN, "2016-04-25"),
        (4f,4f, "2016-05-21"),
        (Float.NaN,Float.NaN, "2016-06-1"),
        (6f,6f, "2016-03-21"))
  ).toDF("id1", "id", "date")

data.na.fill(0).show
+---+---+----------+
|id1| id|      date|
+---+---+----------+
|0.0|0.0| 2016-01-1|
|1.0|1.0| 2016-02-2|
|2.0|2.0|      null|
|0.0|0.0|2016-04-25|
|4.0|4.0|2016-05-21|
|0.0|0.0| 2016-06-1|
|6.0|6.0|2016-03-21|
+---+---+----------+

Answer 2

您可以使用以下表达式将单个数据框列设置为 0 代替 NaN。在此示例中，col1 列中的任何 NaN 值都将替换为 0。

val df = (1 to 10).toDF("col1").withColumn("col1",when(when($"col1".isNull, 0).otherwise($"col1").isNaN, 0).otherwise($"col1"))

Answer 3

我的环境（使用带有 Scala 2.11 的 Spark 2.3.1）没有复制 @ShankarKoirala 的答案 - .na.fill()… 没有捕获 infinity 和 NaN，因为它们不是空值。但是，可以使用 .isin() 函数测试值：

val x1 = Seq((1.0, 1, "a"),(1.0, 1, "a"), (2.0, 2, "b")
           , (Float.NaN, 1, "a"), (Float.PositiveInfinity, 2, "a")
           , (Float.NegativeInfinity, 2, "a"))
        .toDF("Value", "Id", "Name")
x1
  .withColumn("IsItNull", $"Value".isNull)
  .withColumn("IsItBad", $"Value".isin(Double.NaN, Double.PositiveInfinity, Double.NegativeInfinity))
.show()

这将产生以下结果：

+---------+---+----+--------+-------+
|    Value| Id|Name|IsItNull|IsItBad|
+---------+---+----+--------+-------+
|      1.0|  1|   a|   false|  false|
|      1.0|  1|   a|   false|  false|
|      2.0|  2|   b|   false|  false|
|      NaN|  1|   a|   false|   true|
| Infinity|  2|   a|   false|   true|
|-Infinity|  2|   a|   false|   true|
+---------+---+----+--------+-------+

如果需要替换，只需在withColumn() 函数中使用原始列名，并将.isin() 作为when 函数的参数。