Spark / Scala:用最后一个好的观测值填充nan

Question

我正在使用spark 2.0.1，并希望用列中最后一个已知值填充nan值。

spark的唯一参考我可以找到spark/Scala:前向填充最后的观察值或填充null与以前已知的好值与pyspark似乎使用RDD。

我宁愿留在数据帧/数据集的世界，并可能处理多个nan值。这可能吗?

我的假设是数据(最初从例如CSV文件加载)按时间排序，并且该顺序保存在分布式设置中，例如按关闭/最后一个已知值填充是正确的。也许用以前的值填充就足够了，因为对于大多数记录来说，一行中没有2个或更多的nan记录。这真的成立吗?关键是a

myDf.sort("foo").show

将破坏任何顺序，例如，所有null值将首先出现。

一个小例子:

import java.sql.{ Date, Timestamp }
case class FooBar(foo:Date, bar:String)
val myDf = Seq(("2016-01-01","first"),("2016-01-02","second"),("2016-wrongFormat","noValidFormat"), ("2016-01-04","lastAssumingSameDate"))
         .toDF("foo","bar")
         .withColumn("foo", 'foo.cast("Date"))
         .as[FooBar]

结果

+----------+--------------------+
|       foo|                 bar|
+----------+--------------------+
|2016-01-01|               first|
|2016-01-02|              second|
|      null|       noValidFormat|
|2016-01-04|lastAssumingSameDate|
+----------+--------------------+

我想用最后一个已知值来固定值。我怎样才能做到这一点呢?

+----------+--------------------+
|       foo|                 bar|
+----------+--------------------+
|2016-01-01|               first|
|2016-01-02|              second|
|2016-01-02|       noValidFormat|
|2016-01-04|lastAssumingSameDate|
+----------+--------------------+

<标题>编辑

在我的例子中，填充上面一行的值就足够了，因为只有非常有限的错误值。

<标题> edit2 h1> 尝试添加一个索引列

val myDf = Seq(("2016-01-01", "first"), ("2016-01-02", "second"), ("2016-wrongFormat", "noValidFormat"), ("2016-01-04", "lastAssumingSameDate"))
    .toDF("foo", "bar")
    .withColumn("foo", 'foo.cast("Date"))
    .as[FooBar]
    .withColumn("rowId", monotonically_increasing_id())

然后用最后一个值填充。

myDf.withColumn("fooLag", lag('foo, 1) over Window.orderBy('rowId)).show

但是这读出了下面的警告:没有为窗口操作定义分区!将所有数据移动到单个分区，这可能会导致严重的性能下降。如何引入有意义的分区?

+----------+--------------------+-----+----------+
|       foo|                 bar|rowId|    fooLag|
+----------+--------------------+-----+----------+
|2016-01-01|               first|    0|      null|
|2016-01-02|              second|    1|2016-01-01|
|      null|       noValidFormat|    2|2016-01-02|
|2016-01-04|lastAssumingSameDate|    3|      null|
+----------+--------------------+-----+----------+

Answer 1

//用最后一个未知的null填充空字段我试过了，这真的有效!!

val dftxt1 = spark.read.option("header","true").option("sep","t").csv("/sdata/ph/com/r/ph_com_r_ita_javelin/inbound/abc.txt").toDF("line_name", "merge_key", "line_id")
dftxt2.select("line_name","merge_key","line_id").write.mode("overwrite").insertInto("dbname.tablename")
val df = spark.sql("select * from dbname.tablename")
val Df1 = df.withColumn("rowId", monotonically_increasing_id())
import org.apache.spark.sql.expressions.Window
val partitionWindow = Window.orderBy("rowId")
val Df2 = Df1.withColumn("line_id", last("line_id", true) over (partitionWindow))
Df2.show

Answer 2

这是一个中间答案。但是，它不是很好，因为没有分区/只使用单个分区。我还在寻找一个更好的方法来解决这个问题

df
    .withColumn("rowId", monotonically_increasing_id())
    .withColumn("replacement", lag('columnWithNull, 1) over Window.orderBy('rowId))
    .withColumn("columnWithNullReplaced",
      when($"columnWithNull" isNull, "replacement").otherwise($"columnWithNull")
    )

<标题>编辑

我正在使用mapPartitionsWithIndex构建一个更好的解决方案https://gist.github.com/geoHeil/6a23d18ccec085d486165089f9f430f2尚未完成

<标题> edit2 h1>

if (i == 0) {
          lastNotNullRow = toCarryBd.value.get(i + 1).get
        } else {
          lastNotNullRow = toCarryBd.value.get(i - 1).get
        }