Spark / Scala: forward fill z ostatnią obserwacją

Odpowiedź początkowa (założenie pojedynczego szeregu czasowego):

Przede wszystkim unikaj funkcji okien, jeśli nie możesz podać klauzuli PARTITION BY. Przenosi dane na pojedynczą partycję, więc w większości przypadków jest to po prostu niewykonalne.

To, co możesz zrobić, to wypełnić luki na RDD za pomocą mapPartitionsWithIndex. Ponieważ nie podałeś przykładowych danych lub oczekiwanego wyjścia, Uznaj to za pseudokod, a nie prawdziwy program Scala: {]}

• Najpierw zamów DataFrame według daty i przekonwertuj na RDD

  import org.apache.spark.sql.Row
  import org.apache.spark.rdd.RDD
  
  val rows: RDD[Row] = df.orderBy($"Date").rdd
  

• Next lets find the last not null observation per partition

  def notMissing(row: Row): Boolean = ???
  
  val toCarry: scala.collection.Map[Int,Option[org.apache.spark.sql.Row]] = rows
    .mapPartitionsWithIndex{ case (i, iter) => 
      Iterator((i, iter.filter(notMissing(_)).toSeq.lastOption)) }
    .collectAsMap
  

• I przekonwertować to Map na broadcast

  val toCarryBd = sc.broadcast(toCarry)
  

• Wreszcie Mapa nad partycjami po raz kolejny wypełniając luki:

  def fill(i: Int, iter: Iterator[Row]): Iterator[Row] = {
    // If it is the beginning of partition and value is missing
    // extract value to fill from toCarryBd.value
    // Remember to correct for empty / only missing partitions
    // otherwise take last not-null from the current partition
  }
  
  val imputed: RDD[Row] = rows
    .mapPartitionsWithIndex{ case (i, iter) => fill(i, iter) } 
  

• W końcu przekonwertować z powrotem do DataFrame

Edit (partycjonowane / szeregi czasowe na dane grupy):

def fill(iter: List[Row]): List[Row] = {
  // Just go row by row and fill with last non-empty value
  ???
}

val groupedAndSorted = df.rdd
  .groupBy(_.getAs[T]("k"))
  .mapValues(_.toList.sortBy(_.getAs[Int]("Date")))

val rows: RDD[Row] = groupedAndSorted.mapValues(fill).values.flatMap(identity)

val dfFilled = sqlContext.createDataFrame(rows, df.schema)

W ten sposób możesz wypełnić wszystkie kolumny w tym samym czasie.

Czy można to zrobić z DataFrames zamiast konwertować tam iz powrotem do RDD?

To zależy, chociaż jest mało prawdopodobne, aby być skutecznym. Jeśli maksymalna przerwa jest stosunkowo mała, możesz zrobić coś takiego:

import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions.{WindowSpec, Window}
import org.apache.spark.sql.Column

val maxGap: Int = ???  // Maximum gap between observations
val columnsToFill: List[String] = ???  // List of columns to fill
val suffix: String = "_" // To disambiguate between original and imputed 

// Take lag 1 to maxGap and coalesce
def makeCoalesce(w: WindowSpec)(magGap: Int)(suffix: String)(c: String) = {
  // Generate lag values between 1 and maxGap
  val lags = (1 to maxGap).map(lag(col(c), _)over(w))
  // Add current, coalesce and set alias
  coalesce(col(c) +: lags: _*).alias(s"$c$suffix")
}


// For each column you want to fill nulls apply makeCoalesce
val lags: List[Column] = columnsToFill.map(makeCoalesce(w)(maxGap)("_"))


// Finally select
val dfImputed = df.select($"*" :: lags: _*)

Można go łatwo dostosować do użyj innej maksymalnej szczeliny na kolumnę.

Prostszym sposobem osiągnięcia podobnego wyniku w najnowszej wersji Spark jest użycie last z ignoreNulls:

import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions.Window

val w = Window.partitionBy($"k").orderBy($"Date")
  .rowsBetween(Window.unboundedPreceding, -1)

df.withColumn("value", coalesce($"value", last($"value", true).over(w)))

Chociaż możliwe jest porzucenie klauzuli partitionBy i zastosowanie tej metody globalnie, byłoby to zbyt kosztowne przy dużych zbiorach danych.