<jupyter_start><jupyter_text>Working with Window Functions in SparkIn this tutorial, we will work with Window function in Spark. Window Functionswere introduced in Spark v1.4 and are explained in this awesome Databricks[blog post](https://databricks.com/blog/2015/07/15/introducing-window-functions-in-spark-sql.html).For this tutorial, we will use the following sample data:<jupyter_code>wdf = spark.createDataFrame([
  (1234, '2018-03-01', 40),
  (1234, '2018-03-15', 25),
  (1234, '2018-04-01', 25),
  (1234, '2018-04-30', 30),
  (4567, '2018-03-01', 200),
  (4567, '2018-03-13', 221),
  (4567, '2018-04-01', 700),
  (4567, '2018-04-06', 400),
  (4567, '2018-04-29', 500),
  (9999, '2018-03-01', 150),
  (8888, '2018-03-01', 15),
  (8888, '2018-03-30', 20)
]).toDF('id', 'start', 'nb')<jupyter_output><empty_output><jupyter_text>To work with SQL, let's create a temporary view over our dataframe. A viewcan be used exactly as a table (i.e. Hive Table).<jupyter_code>wdf.createOrReplaceTempView('wdftable')<jupyter_output><empty_output><jupyter_text>In the following we will create windows over the dataframe partitioned by `id`column.  Define a WindowA Window over a dataframe can be defined as follows:<jupyter_code>from pyspark.sql import Window
## create a window partitioned by id and order windows data by start column
win = Window.partitionBy('id').orderBy(wdf.start.desc())<jupyter_output><empty_output><jupyter_text>Window OperationsIn the following we will review a large set of window functions.  Rank`pyspark.sql.functions.rank()` is a window function that returns the rank ofrows within a window partition.<jupyter_code>from pyspark.sql import functions as F
rankfct = F.rank().over(win)
rankdf = wdf.select('*', rankfct.alias('rank'))

rankdf.show()<jupyter_output>+----+----------+---+----+
|  id|     start| nb|rank|
+----+----------+---+----+
|8888|2018-03-30| 20|   1|
|8888|2018-03-01| 15|   2|
|9999|2018-03-01|150|   1|
|1234|2018-04-30| 30|   1|
|1234|2018-04-01| 25|   2|
|1234|2018-03-15| 25|   3|
|1234|2018-03-01| 40|   4|
|4567|2018-04-29|500|   1|
|4567|2018-04-06|400|   2|
|4567|2018-04-01|700|   3|
|4567|2018-03-13|221|   4|
|4567|2018-03-01|200|   5|
+----+----------+---+----+<jupyter_text>In `SQL`, the same operation can be achieved with:<jupyter_code>rankdf = spark.sql("SELECT id, start, nb, RANK() OVER (PARTITION BY id ORDER BY start desc) as rank FROM wdftable")

rankdf.show()<jupyter_output>+----+----------+---+----+
|  id|     start| nb|rank|
+----+----------+---+----+
|8888|2018-03-30| 20|   1|
|8888|2018-03-01| 15|   2|
|9999|2018-03-01|150|   1|
|1234|2018-04-30| 30|   1|
|1234|2018-04-01| 25|   2|
|1234|2018-03-15| 25|   3|
|1234|2018-03-01| 40|   4|
|4567|2018-04-29|500|   1|
|4567|2018-04-06|400|   2|
|4567|2018-04-01|700|   3|
|4567|2018-03-13|221|   4|
|4567|2018-03-01|200|   5|
+----+----------+---+----+<jupyter_text>SumWhen using the `sum` function over a window, we will have the cumulative `sum`as we move over the Window. Check the results given by this sample code:<jupyter_code>cumulsum = F.sum(wdf.nb).over(win)

sumdf = wdf.select('*', cumulsum.alias('sum'))

sumdf.show()<jupyter_output>+----+----------+---+----+
|  id|     start| nb| sum|
+----+----------+---+----+
|8888|2018-03-30| 20|  20|
|8888|2018-03-01| 15|  35|
|9999|2018-03-01|150| 150|
|1234|2018-04-30| 30|  30|
|1234|2018-04-01| 25|  55|
|1234|2018-03-15| 25|  80|
|1234|2018-03-01| 40| 120|
|4567|2018-04-29|500| 500|
|4567|2018-04-06|400| 900|
|4567|2018-04-01|700|1600|
|4567|2018-03-13|221|1821|
|4567|2018-03-01|200|2021|
+----+----------+---+----+<jupyter_text>In `SQL`, the same operation can be achieved with:<jupyter_code>sumdf = spark.sql("SELECT id, start, nb, sum(nb) OVER (PARTITION BY id ORDER BY start desc) as sum FROM wdftable")

sumdf.show()<jupyter_output>+----+----------+---+----+
|  id|     start| nb| sum|
+----+----------+---+----+
|8888|2018-03-30| 20|  20|
|8888|2018-03-01| 15|  35|
|9999|2018-03-01|150| 150|
|1234|2018-04-30| 30|  30|
|1234|2018-04-01| 25|  55|
|1234|2018-03-15| 25|  80|
|1234|2018-03-01| 40| 120|
|4567|2018-04-29|500| 500|
|4567|2018-04-06|400| 900|
|4567|2018-04-01|700|1600|
|4567|2018-03-13|221|1821|
|4567|2018-03-01|200|2021|
+----+----------+---+----+<jupyter_text>Lead`lead` is a window function that returns the value that is one (by default) rowafter the current row.<jupyter_code>leadfct = F.lead(wdf.start, 1).over(win)
leaddf = wdf.select('*', leadfct.alias('previous_start'))

leaddf.show()<jupyter_output>+----+----------+---+--------------+
|  id|     start| nb|previous_start|
+----+----------+---+--------------+
|8888|2018-03-30| 20|    2018-03-01|
|8888|2018-03-01| 15|          null|
|9999|2018-03-01|150|          null|
|1234|2018-04-30| 30|    2018-04-01|
|1234|2018-04-01| 25|    2018-03-15|
|1234|2018-03-15| 25|    2018-03-01|
|1234|2018-03-01| 40|          null|
|4567|2018-04-29|500|    2018-04-06|
|4567|2018-04-06|400|    2018-04-01|
|4567|2018-04-01|700|    2018-03-13|
|4567|2018-03-13|221|    2018-03-01|
|4567|2018-03-01|200|          null|
+----+----------+---+--------------+<jupyter_text>In `SQL`, the same operation can be achieved with:<jupyter_code>leaddf = spark.sql('SELECT id, start, nb, LEAD(start) OVER (PARTITION BY id ORDER BY start DESC) as previous_start from wdftable')

leaddf.show()<jupyter_output>+----+----------+---+--------------+
|  id|     start| nb|previous_start|
+----+----------+---+--------------+
|8888|2018-03-30| 20|    2018-03-01|
|8888|2018-03-01| 15|          null|
|9999|2018-03-01|150|          null|
|1234|2018-04-30| 30|    2018-04-01|
|1234|2018-04-01| 25|    2018-03-15|
|1234|2018-03-15| 25|    2018-03-01|
|1234|2018-03-01| 40|          null|
|4567|2018-04-29|500|    2018-04-06|
|4567|2018-04-06|400|    2018-04-01|
|4567|2018-04-01|700|    2018-03-13|
|4567|2018-03-13|221|    2018-03-01|
|4567|2018-03-01|200|          null|
+----+----------+---+--------------+<jupyter_text>Lag`lag` is a window function that returns the value that is one (by default) rowbefore the current row.<jupyter_code>lagfct = F.lag(wdf.start, 1).over(win)
lagdf = wdf.select('*', lagfct.alias('end'))

lagdf.show()<jupyter_output>+----+----------+---+----------+
|  id|     start| nb|       end|
+----+----------+---+----------+
|8888|2018-03-30| 20|      null|
|8888|2018-03-01| 15|2018-03-30|
|9999|2018-03-01|150|      null|
|1234|2018-04-30| 30|      null|
|1234|2018-04-01| 25|2018-04-30|
|1234|2018-03-15| 25|2018-04-01|
|1234|2018-03-01| 40|2018-03-15|
|4567|2018-04-29|500|      null|
|4567|2018-04-06|400|2018-04-29|
|4567|2018-04-01|700|2018-04-06|
|4567|2018-03-13|221|2018-04-01|
|4567|2018-03-01|200|2018-03-13|
+----+----------+---+----------+<jupyter_text>In `SQL`, the same operation can be achieved with:<jupyter_code>lagdf = spark.sql('SELECT id, start, nb, LAG(start) OVER (PARTITION BY id ORDER BY start DESC) as end from wdftable')

lagdf.show()<jupyter_output>+----+----------+---+----------+
|  id|     start| nb|       end|
+----+----------+---+----------+
|8888|2018-03-30| 20|      null|
|8888|2018-03-01| 15|2018-03-30|
|9999|2018-03-01|150|      null|
|1234|2018-04-30| 30|      null|
|1234|2018-04-01| 25|2018-04-30|
|1234|2018-03-15| 25|2018-04-01|
|1234|2018-03-01| 40|2018-03-15|
|4567|2018-04-29|500|      null|
|4567|2018-04-06|400|2018-04-29|
|4567|2018-04-01|700|2018-04-06|
|4567|2018-03-13|221|2018-04-01|
|4567|2018-03-01|200|2018-03-13|
+----+----------+---+----------+<jupyter_text>First`first` returns the first value in a window partition.<jupyter_code>firstfct = F.first(wdf.start).over(win)
firstdf = wdf.select('*', firstfct.alias('first'))

firstdf.show()<jupyter_output>+----+----------+---+----------+
|  id|     start| nb|     first|
+----+----------+---+----------+
|8888|2018-03-30| 20|2018-03-30|
|8888|2018-03-01| 15|2018-03-30|
|9999|2018-03-01|150|2018-03-01|
|1234|2018-04-30| 30|2018-04-30|
|1234|2018-04-01| 25|2018-04-30|
|1234|2018-03-15| 25|2018-04-30|
|1234|2018-03-01| 40|2018-04-30|
|4567|2018-04-29|500|2018-04-29|
|4567|2018-04-06|400|2018-04-29|
|4567|2018-04-01|700|2018-04-29|
|4567|2018-03-13|221|2018-04-29|
|4567|2018-03-01|200|2018-04-29|
+----+----------+---+----------+<jupyter_text>In `SQL`, the same operation can be achieved with:<jupyter_code>firstdf = spark.sql('SELECT id, start, nb, FIRST(start) OVER (PARTITION BY id ORDER BY start DESC) as first from wdftable')

firstdf.show()<jupyter_output>+----+----------+---+----------+
|  id|     start| nb|     first|
+----+----------+---+----------+
|8888|2018-03-30| 20|2018-03-30|
|8888|2018-03-01| 15|2018-03-30|
|9999|2018-03-01|150|2018-03-01|
|1234|2018-04-30| 30|2018-04-30|
|1234|2018-04-01| 25|2018-04-30|
|1234|2018-03-15| 25|2018-04-30|
|1234|2018-03-01| 40|2018-04-30|
|4567|2018-04-29|500|2018-04-29|
|4567|2018-04-06|400|2018-04-29|
|4567|2018-04-01|700|2018-04-29|
|4567|2018-03-13|221|2018-04-29|
|4567|2018-03-01|200|2018-04-29|
+----+----------+---+----------+<jupyter_text>Last`last` returns the last value in a window partition.<jupyter_code>lastfct = F.last(wdf.start).over(win)
lastdf = wdf.select('*', lastfct.alias('last_not_really'))

lastdf.show()<jupyter_output>+----+----------+---+---------------+
|  id|     start| nb|last_not_really|
+----+----------+---+---------------+
|8888|2018-03-30| 20|     2018-03-30|
|8888|2018-03-01| 15|     2018-03-01|
|9999|2018-03-01|150|     2018-03-01|
|1234|2018-04-30| 30|     2018-04-30|
|1234|2018-04-01| 25|     2018-04-01|
|1234|2018-03-15| 25|     2018-03-15|
|1234|2018-03-01| 40|     2018-03-01|
|4567|2018-04-29|500|     2018-04-29|
|4567|2018-04-06|400|     2018-04-06|
|4567|2018-04-01|700|     2018-04-01|
|4567|2018-03-13|221|     2018-03-13|
|4567|2018-03-01|200|     2018-03-01|
+----+----------+---+---------------+<jupyter_text>In `SQL`, the same operation can be achieved with:<jupyter_code>lastdf = spark.sql('SELECT id, start, nb, LAST(start) OVER (PARTITION BY id ORDER BY start DESC) as last_not_really from wdftable')

lastdf.show()<jupyter_output>+----+----------+---+---------------+
|  id|     start| nb|last_not_really|
+----+----------+---+---------------+
|8888|2018-03-30| 20|     2018-03-30|
|8888|2018-03-01| 15|     2018-03-01|
|9999|2018-03-01|150|     2018-03-01|
|1234|2018-04-30| 30|     2018-04-30|
|1234|2018-04-01| 25|     2018-04-01|
|1234|2018-03-15| 25|     2018-03-15|
|1234|2018-03-01| 40|     2018-03-01|
|4567|2018-04-29|500|     2018-04-29|
|4567|2018-04-06|400|     2018-04-06|
|4567|2018-04-01|700|     2018-04-01|
|4567|2018-03-13|221|     2018-03-13|
|4567|2018-03-01|200|     2018-03-01|
+----+----------+---+---------------+<jupyter_text>As you can see in the results, the newly created column by `last` window functiondo not contain what we really expected. It contains exactly the same values asthe content of `start` column. This is however normal, the reason is `range` ofthe Window being by default from [`Window.unboundedPreceding`, `Window.currentRow`]which practically means from all preceding row to the current row. Using such rangeit is not surprising that the `last` row will always be the current row.In order to consider the complete range of the window partitions, we need tospecify the range:<jupyter_code>## range from all preceding rows to all following rows
win = Window.partitionBy('id').orderBy(wdf.start.desc()).rowsBetween(-Window.unboundedFollowing, Window.unboundedFollowing)

lastfct = F.last(wdf.start).over(win)

lastdf = wdf.select('*', lastfct.alias('last'))

lastdf.show()<jupyter_output>+----+----------+---+----------+
|  id|     start| nb|      last|
+----+----------+---+----------+
|8888|2018-03-30| 20|2018-03-01|
|8888|2018-03-01| 15|2018-03-01|
|9999|2018-03-01|150|2018-03-01|
|1234|2018-04-30| 30|2018-03-01|
|1234|2018-04-01| 25|2018-03-01|
|1234|2018-03-15| 25|2018-03-01|
|1234|2018-03-01| 40|2018-03-01|
|4567|2018-04-29|500|2018-03-01|
|4567|2018-04-06|400|2018-03-01|
|4567|2018-04-01|700|2018-03-01|
|4567|2018-03-13|221|2018-03-01|
|4567|2018-03-01|200|2018-03-01|
+----+----------+---+----------+<jupyter_text>In `SQL`, the same operation can be achieved with:<jupyter_code>lastdf = spark.sql('SELECT id, start, nb, LAST(start) OVER (PARTITION BY id ORDER BY start DESC ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING) as last from wdftable')

lastdf.show()<jupyter_output>+----+----------+---+----------+
|  id|     start| nb|      last|
+----+----------+---+----------+
|8888|2018-03-30| 20|2018-03-01|
|8888|2018-03-01| 15|2018-03-01|
|9999|2018-03-01|150|2018-03-01|
|1234|2018-04-30| 30|2018-03-01|
|1234|2018-04-01| 25|2018-03-01|
|1234|2018-03-15| 25|2018-03-01|
|1234|2018-03-01| 40|2018-03-01|
|4567|2018-04-29|500|2018-03-01|
|4567|2018-04-06|400|2018-03-01|
|4567|2018-04-01|700|2018-03-01|
|4567|2018-03-13|221|2018-03-01|
|4567|2018-03-01|200|2018-03-01|
+----+----------+---+----------+<jupyter_text>> It is critical to correctly specify the range when defining a Window. Theresults highly depend on this range as we have seen with `last` function.  Max`max` returns the maximum value in the window partition range.Using the comments from last section, add a new column to the dataframe containingthe maximum value of `nb` column in the window partition.<jupyter_code>## Add max<jupyter_output><empty_output><jupyter_text>Add another column containing the difference between the maximum value over thewindow and the current value of the row. See how the window function can effectivelybe an expression.<jupyter_code>## max - current value<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.2585607573404334, 'fraction_numerical': 0.26225870867539386, 'mean_word_length': 2.98408956982911, 'pattern_counts': {'":': 0, '<': 63, '<?xml version=': 0, '>': 64, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '19', 'dir': 'jupyter-structured-clean-dedup', 'id': '431013', 'license': 'MIT', 'n_tokens_mistral': 7899, 'n_tokens_neox': 5569, 'n_words': 1081, 'path': 'tutorials/working_with_window_functions.ipynb', 'repo_name': 'bachwehbi/tutorials', 'size': '13521'}