Window Functions (aka Analytic Functions) in Spark.
As of Spark 1.4.0 we now have support for window functions (aka analytic functions) in SparkSQL. At Sonra we are heavy users of SparkSQL to handle data transformations for structured data. We also use it in combination with cached RDDs and Tableau for business intelligence and visual analytics.
[big_data_promotion]
Spark SQL and Window Functions: The rationale
I am a strong supporter of using SQL for performing ETL and data transformations of structured data. There are several reasons for this:
- A lot of people know SQL. Similar to English in the real world it is the lingua franca of data.
- Most requirements can be met with SQL for structured data. Admittedly, it is not so good when it comes to querying graphs, complex trees, non-structured data, and complex many to many relationships.
- In a lot of cases you have to write less code using the SparkSQL abstraction layer.
Window functions allow you to perform inter-row calculations that in traditional SQL require ugly and costly self-joins, e.g. a moving average can be calculated in one line of code. If you want to learn more about window functions in SQL I would recommend the SQL Cookbook from O’Reilly.
An example
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 |
import org.apache.spark.rdd.RDD import org.apache.spark.sql.catalyst.expressions.Row import org.apache.spark.sql.catalyst.types._ import org.apache.spark.sql.hive.HiveContext import org.apache.spark.{SparkConf, SparkContext} object WindowingQueryTest { def main(args: Array[String]) { val sc = new SparkContext(new SparkConf().setAppName("WndFnc_demo")) val hiveContext = new HiveContext(sc) val data = Seq(("A", 4), ("C", 1), ("D", 1), ("B", 2), ("B", 2), ("D", 4), ("A", 1), ("B", 4)) val withRowNumbers: Seq[(String, Int, Int)] = data.zipWithIndex.map(e => (e._1._1, e._1._2, e._2)) val rdd: RDD[Row] = sc.parallelize(withRowNumbers).map(triplet => Row(triplet._1, triplet._2, triplet._3)) hiveContext.sql("DROP TABLE IF EXISTS delme") // hiveContext.applySchema(rdd, schema).saveAsTable("delme") hiveContext.sql( """CREATE TABLE `delme`( `key` string, `val` int, `ord` int)""") val schema = StructType(Seq(StructField("key", StringType), StructField("val", IntegerType), StructField("ord", IntegerType))) hiveContext.applySchema(rdd, schema).insertInto("delme") val qRes = hiveContext.sql("""SELECT key, val ,MAX(val)OVER(PARTITION BY key) mx ,MIN(val)OVER(PARTITION BY key) mn ,row_number() OVER(ORDER BY ord desc) revord ,rank() OVER(ORDER BY val) rnk FROM delme""") qRes.collect().foreach(println) } } +------+------+-----+-----+---------+------+ | key | val | mx | mn | revord | rnk | +------+------+-----+-----+---------+------+ | A | 1 | 1 | 1 | 2 | 1 | | A | 1 | 1 | 1 | 9 | 1 | | B | 2 | 4 | 2 | 4 | 5 | | B | 2 | 4 | 2 | 5 | 5 | | B | 4 | 4 | 2 | 1 | 7 | | B | 4 | 4 | 2 | 8 | 7 | | C | 1 | 1 | 1 | 7 | 1 | | D | 1 | 4 | 1 | 6 | 1 | | D | 4 | 4 | 1 | 3 | 7 | +------+------+-----+-----+---------+------+ |
Merging the patch into older Spark versions
We are using version 1.3.1 of Spark on top of the MapR Hadoop distribution. As Spark 1.3.1 does not ship with the Window functions we merged the 1.4.0 patch into the 1.3.0 MapR Spark version. You can contact us for the assembly if you want to achieve the same (also for Spark 1.2.1),
