<jupyter_start><jupyter_text>Using first and last functionsLet us understand the usage of first and last value functions. Let us start spark context for this Notebook so that we can execute the code provided. You can sign up for our [10 node state of the art cluster/labs](https://labs.itversity.com/plans) to learn Spark SQL using our unique integrated LMS.<jupyter_code>from pyspark.sql import SparkSession

import getpass
username = getpass.getuser()

spark = SparkSession. \
    builder. \
    config('spark.ui.port', '0'). \
    config("spark.sql.warehouse.dir", f"/user/{username}/warehouse"). \
    enableHiveSupport(). \
    appName(f'{username} | Python - Windowing Functions'). \
    master('yarn'). \
    getOrCreate()<jupyter_output><empty_output><jupyter_text>If you are going to use CLIs, you can use Spark SQL using one of the 3 approaches.**Using Spark SQL**```spark2-sql \    --master yarn \    --conf spark.ui.port=0 \    --conf spark.sql.warehouse.dir=/user/${USER}/warehouse```**Using Scala**```spark2-shell \    --master yarn \    --conf spark.ui.port=0 \    --conf spark.sql.warehouse.dir=/user/${USER}/warehouse```**Using Pyspark**```pyspark2 \    --master yarn \    --conf spark.ui.port=0 \    --conf spark.sql.warehouse.dir=/user/${USER}/warehouse```<jupyter_code>spark.conf.set('spark.sql.shuffle.partitions', '2')<jupyter_output><empty_output><jupyter_text>* Let us get highest paid employee and least paid employee with in each department for each employee using employees data set.* You can also use max to get max salary for each department, but you cannot get other attributes related to ma salary such as employee id, name etc. With first or last, you can get other details as well.<jupyter_code>employeesPath = '/public/hr_db/employees'
employees = spark. \
    read. \
    format('csv'). \
    option('sep', '\t'). \
    schema('''employee_id INT, 
              first_name STRING, 
              last_name STRING, 
              email STRING,
              phone_number STRING, 
              hire_date STRING, 
              job_id STRING, 
              salary FLOAT,
              commission_pct STRING,
              manager_id STRING, 
              department_id STRING
            '''). \
    load(employeesPath)
from pyspark.sql.functions import col
employees. \
    select('employee_id', 
           col('department_id').cast('int').alias('department_id'), 
           'salary'
          ). \
    orderBy('department_id', 'salary'). \
    show()
from pyspark.sql.window import Window
spec = Window. \
    partitionBy('department_id'). \
    orderBy(col('salary').desc())
from pyspark.sql.functions import first
employees. \
    select('employee_id', 
           col('department_id').cast('int').alias('department_id'), 
           'salary'
          ). \
    withColumn("highest_salary", first('salary').over(spec)). \
    withColumn("highest_employee_id", first('employee_id').over(spec)). \
    orderBy("department_id", col("salary").desc()). \
    show()<jupyter_output>+-----------+-------------+-------+--------------+-------------------+
|employee_id|department_id| salary|highest_salary|highest_employee_id|
+-----------+-------------+-------+--------------+-------------------+
|        178|         null| 7000.0|        7000.0|                178|
|        200|           10| 4400.0|        4400.0|                200|
|        201|           20|13000.0|       13000.0|                201|
|        202|           20| 6000.0|       13000.0|                201|
|        114|           30|11000.0|       11000.0|                114|
|        115|           30| 3100.0|       11000.0|                114|
|        116|           30| 2900.0|       11000.0|                114|
|        117|           30| 2800.0|       11000.0|                114|
|        118|           30| 2600.0|       11000.0|                114|
|        119|           30| 2500.0|       11000.0|                114|
|        203|           40| 6500.0|        6500.0|                203|
|     [...]<jupyter_text>* The default functionality of last function is to use the rows between unbounded preceding to current row. We need to change the rows between to unbounded preceding to unbounded following.<jupyter_code>from pyspark.sql.functions import last
spec = Window. \
    partitionBy('department_id'). \
    orderBy(col('salary').desc())
employees. \
    select('employee_id', 
           col('department_id').cast('int').alias('department_id'), 
           'salary'
          ). \
    withColumn("highest_salary", last('salary').over(spec)). \
    withColumn("highest_employee_id", last('employee_id').over(spec)). \
    orderBy("department_id", col("salary").desc()). \
    show()
help(spec)
spec = Window. \
    partitionBy('department_id'). \
    orderBy(col('salary')). \
    rowsBetween(Window.unboundedPreceding, Window.currentRow)
employees. \
    select('employee_id', 
           col('department_id').cast('int').alias('department_id'), 
           'salary'
          ). \
    withColumn("highest_salary", last('salary').over(spec)). \
    withColumn("highest_employee_id", last('employee_id').over(spec)). \
    orderBy("department_id", col("salary").desc()). \
    show()<jupyter_output>+-----------+-------------+-------+--------------+-------------------+
|employee_id|department_id| salary|highest_salary|highest_employee_id|
+-----------+-------------+-------+--------------+-------------------+
|        178|         null| 7000.0|        7000.0|                178|
|        200|           10| 4400.0|        4400.0|                200|
|        201|           20|13000.0|       13000.0|                201|
|        202|           20| 6000.0|        6000.0|                202|
|        114|           30|11000.0|       11000.0|                114|
|        115|           30| 3100.0|        3100.0|                115|
|        116|           30| 2900.0|        2900.0|                116|
|        117|           30| 2800.0|        2800.0|                117|
|        118|           30| 2600.0|        2600.0|                118|
|        119|           30| 2500.0|        2500.0|                119|
|        203|           40| 6500.0|        6500.0|                203|
|     [...]
{'fraction_non_alphanumeric': 0.17188252286952335, 'fraction_numerical': 0.06692344727973037, 'mean_word_length': 1.6827378390012915, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 15, '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': '4', 'dir': 'jupyter-structured-clean-dedup', 'id': '589761', 'license': 'MIT', 'n_tokens_mistral': 2165, 'n_tokens_neox': 1819, 'n_words': 416, 'path': '07_windowing_functions/07_using_first_and_last_functions.ipynb', 'repo_name': 'itversity/pyspark', 'size': '6231'}