Spark / Scala: remplissage en avant avec la dernière observation

Réponse initiale (hypothèse d'une seule série temporelle):

Essayez D'abord d'éviter les fonctions de fenêtre si vous ne pouvez pas fournir la clause PARTITION BY. Il déplace les données vers une seule partition, donc la plupart du temps, ce n'est tout simplement pas faisable.

Ce que vous pouvez faire est de combler les lacunes sur RDD utiliser mapPartitionsWithIndex. Puisque vous n'avez pas fourni d'exemple de données ou de sortie attendue, considérez ceci comme un pseudocode pas un vrai programme Scala:

• Permet D'abord de commander DataFrame par date et de convertir en RDD

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

• Suivant permet de trouver la dernière observation non nulle par 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
  

• Et convertissez ceci Map en diffusion

  val toCarryBd = sc.broadcast(toCarry)
  

• Enfin mapper sur les partitions une fois de plus combler les lacunes:

  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) } 
  

• Enfin, convertissez en DataFrame

Modifier (partitionné / série temporelle par données de groupe):

Le diable est dans les détails. Si vos données sont partitionnées après tout alors un ensemble le problème peut être résolu en utilisant groupBy. Supposons que vous partitionnez simplement par colonne " v " de type T et Date est un horodatage entier:

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)

De Cette façon, vous pouvez remplir toutes les colonnes en même temps.

Cela peut-il être fait avec des DataFrames au lieu de convertir des va-et-vient en RDD?

Cela dépend, bien qu'il soit peu probable d'être efficace. Si l'écart maximum est relativement faible, vous pouvez faire quelque chose comme ceci:

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: _*)

Il peut être facilement ajusté à utilisez un écart maximum différent par colonne.

Un moyen plus simple d'obtenir un résultat similaire dans la dernière version de Spark est d'utiliser last avec 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)))

Bien qu'il soit possible de supprimer la clause partitionBy et d'appliquer cette méthode globalement, cela coûterait prohibitif avec de grands ensembles de données.