A data engineer is working with a PySpark DataFrame `sensor_data` that contains `device_id`, `reading_time`, and `temperature`. They need to calculate the difference in temperature between the current reading and the previous reading for each device. If there is no previous reading for a device (i.e., it's the first reading), the difference should be NULL. Which PySpark window function combination should be used?
- Awindow_spec = Window.partitionBy('device_id').orderBy('reading_time') df.withColumn('temp_diff', col('temperature') - first_value('temperature').over(window_spec))
- Bwindow_spec = Window.partitionBy('device_id').orderBy('reading_time') df.withColumn('temp_diff', col('temperature') - col('temperature').shift(1).over(window_spec))
- Cwindow_spec = Window.partitionBy('device_id').orderBy('reading_time') df.withColumn('temp_diff', col('temperature') - lead('temperature', 1).over(window_spec))
- Dwindow_spec = Window.partitionBy('device_id').orderBy('reading_time') df.withColumn('temp_diff', col('temperature') - lag('temperature', 1).over(window_spec))
Show answer & explanationAnswer & explanation
Correct answer: D. window_spec = Window.partitionBy('device_id').orderBy('reading_time') df.withColumn('temp_diff', col('temperature') - lag('temperature', 1).over(window_spec))
To calculate the difference from the previous row, the `lag()` window function is appropriate. It retrieves the value from a preceding row within the defined window. The `Window.partitionBy('device_id').orderBy('reading_time')` correctly sets up the window to operate independently for each device, ordered by time, ensuring 'previous' refers to the chronologically prior reading for that specific device.
Why the other options are wrong
- A. FIRST_VALUE() would return the very first temperature reading for the device, not the immediate previous one.
- B. There is no `shift()` method directly on a column expression in PySpark for windowing in this manner. `shift()` is typically a Pandas DataFrame method.
- C. LEAD() would get the *next* temperature reading, not the previous one.
PySpark LAG() Window Function
A PySpark window function that retrieves the value of an expression from a row that precedes the current row by a specified offset within its partition, ordered by a given column.
- Used to compare current row data with previous row data.
- Requires `Window.partitionBy()` and `orderBy()` for definition.
- Returns `NULL` by default if no preceding row is found.
- Essential for time-series analysis like calculating deltas.
Memory trick: Lag back one step, for each device's ordered path.