Funke und SparkSQL: Wie zu imitieren-Fenster-Funktion?

Beschreibung

Gegeben ein dataframe df

id |       date
---------------
 1 | 2015-09-01
 2 | 2015-09-01
 1 | 2015-09-03
 1 | 2015-09-04
 2 | 2015-09-04

Ich möchte erstellen Sie eine laufende Zähler oder index,

  • gruppiert, die von der gleichen id und
  • nach Datum sortiert in einer Gruppe,

damit

id |       date |  counter
--------------------------
 1 | 2015-09-01 |        1
 1 | 2015-09-03 |        2
 1 | 2015-09-04 |        3
 2 | 2015-09-01 |        1
 2 | 2015-09-04 |        2

Dies ist etwas, das ich erreichen kann mit der Fenster-Funktion, z.B.

val w = Window.partitionBy("id").orderBy("date")
val resultDF = df.select( df("id"), rowNumber().over(w) )

Leider Spark 1.4.1 nicht unterstützt window-Funktionen für regelmäßige dataframes:

org.apache.spark.sql.AnalysisException: Could not resolve window function 'row_number'. Note that, using window functions currently requires a HiveContext;

Fragen

  • Wie kann ich das erreichen der obigen Berechnung auf die aktuelle Spark 1.4.1 ohne mit dem Fenster funktioniert?
  • Wann wird der window-Funktion für die regelmäßige dataframes werden unterstützt in der Funke?

Dank!

  • Tun Sie verwenden müssen, dataframes & SQL -, oder könnten Sie verwenden RDDs? Dies ist ganz einfach mit der groupBy-Methode.
  • RDDs wäre auch in Ordnung. Können Sie bitte skizzieren Sie Ihre Idee mit einer kleinen code-Auszug? Als der SparkSQL ich sehe momentan keinen Weg, wie dies zu tun: haben Sie eine Idee?
InformationsquelleAutor Martin Senne | 2015-09-04
  1. 6

    Können Sie dies mit RDDs. Persönlich finde ich die API für RDDs macht viel mehr Sinn - ich weiß nicht immer wollen, dass meine Daten zu "flach", wie ein dataframe.

    val df = sqlContext.sql("select 1, '2015-09-01'"
    ).unionAll(sqlContext.sql("select 2, '2015-09-01'")
    ).unionAll(sqlContext.sql("select 1, '2015-09-03'")
    ).unionAll(sqlContext.sql("select 1, '2015-09-04'")
    ).unionAll(sqlContext.sql("select 2, '2015-09-04'"))

//dataframe as an RDD (of Row objects)
df.rdd 
  //grouping by the first column of the row
  .groupBy(r => r(0)) 
  //map each group - an Iterable[Row] - to a list and sort by the second column
  .map(g => g._2.toList.sortBy(row => row(1).toString))     
  .collect()

    Den oben ergeben sich wie folgt:

    Array[List[org.apache.spark.sql.Row]] = 
Array(
  List([1,2015-09-01], [1,2015-09-03], [1,2015-09-04]), 
  List([2,2015-09-01], [2,2015-09-04]))

    Wenn Sie möchten, dass die position innerhalb der "Gruppe" als gut, können Sie verwenden zipWithIndex.

    df.rdd.groupBy(r => r(0)).map(g => 
    g._2.toList.sortBy(row => row(1).toString).zipWithIndex).collect()

Array[List[(org.apache.spark.sql.Row, Int)]] = Array(
  List(([1,2015-09-01],0), ([1,2015-09-03],1), ([1,2015-09-04],2)),
  List(([2,2015-09-01],0), ([2,2015-09-04],1)))

    Du könnte glätten Sie diese zurück, um eine einfache Liste/Array von Row Objekte mit FlatMap, aber wenn Sie brauchen, um alles auf die 'Gruppe', die nicht eine große Idee sein.

    Der Nachteil bei der Verwendung der RDD ist, dass es mühsam konvertieren von DataFrame zu RDD und wieder zurück.

    • Vielen Dank!!! Das war die Lösung, die ich suchte. Hmm, ich war einfach nicht "mutig" genug, die regelmäßige Scala list Operationen, sobald die groupBy gemacht wurde ....
    • was passiert, wenn meine "g._2.toList.sortBy" - Liste hat Millionen von Elementen, ich kann Sie abholen
    InformationsquelleAutor Kirk Broadhurst
  2. 7

    Können Sie HiveContext für lokale DataFrames so gut und, es sei denn, Sie haben einen sehr guten Grund, nicht zu, ist es wahrscheinlich eine gute Idee, sowieso. Es ist ein Standard - SQLContext erhältlich in spark-shell und pyspark shell (wie jetzt sparkR scheint einfach SQLContext) und dessen parser-empfohlen von Spark SQL und DataFrame Guide.

    import org.apache.spark.{SparkContext, SparkConf}
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.rowNumber

object HiveContextTest {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("Hive Context")
    val sc = new SparkContext(conf)
    val sqlContext = new HiveContext(sc)
    import sqlContext.implicits._

    val df = sc.parallelize(
        ("foo", 1) :: ("foo", 2) :: ("bar", 1) :: ("bar", 2) :: Nil
    ).toDF("k", "v")

    val w = Window.partitionBy($"k").orderBy($"v")
    df.select($"k", $"v", rowNumber.over(w).alias("rn")).show
  }
}
    InformationsquelleAutor zero323
  3. 3

    Ich bin völlig einverstanden, dass die Fenster-Funktionen für DataFrames sind der Weg zu gehen, wenn Sie Spark version (>=)1.5. Aber wenn Sie wirklich stecken, die mit einer älteren version(e.g 1.4.1), hier ist ein hacky Weg, um dieses Problem zu lösen

    val df = sc.parallelize((1, "2015-09-01") :: (2, "2015-09-01") :: (1, "2015-09-03") :: (1, "2015-09-04") :: (1, "2015-09-04") :: Nil)
           .toDF("id", "date")

val dfDuplicate = df.selecExpr("id as idDup", "date as dateDup")
val dfWithCounter = df.join(dfDuplicate,$"id"===$"idDup")
                      .where($"date"<=$"dateDup")
                      .groupBy($"id", $"date")
                      .agg($"id", $"date", count($"idDup").as("counter"))
                      .select($"id",$"date",$"counter")

    Nun, wenn Sie dfWithCounter.show

    Erhalten Sie:

    +---+----------+-------+                                                        
| id|      date|counter|
+---+----------+-------+
|  1|2015-09-01|      1|
|  1|2015-09-04|      3|
|  1|2015-09-03|      2|
|  2|2015-09-01|      1|
|  2|2015-09-04|      2|
+---+----------+-------+

    Beachten Sie, dass date ist nicht sortiert, aber die counter korrekt ist. Sie können auch ändern, die Reihenfolge der counter durch ändern der <= zu >= im where - Anweisung.

    InformationsquelleAutor Sayon M
Schreibe einen Kommentar