I have a dataframe with columns like these:
df_lp.show()
+--------------------+-------------+--------------------+------+
| ts| uid| pid|action|
+--------------------+-------------+--------------------+------+
|2017-03-28 09:34:...|1950269663250|IST334-0149064968...| <L|
|2017-03-31 05:50:...|S578448696405|IST334-0149089179...| <L|
|2017-03-28 09:38:...|1950269663250|IST334-0149064968...| <L|
|2017-03-30 09:26:...| 412310802992|IST334-1212011845...| <L|
ts is a timestamp, all the other columns are strings. action can take one among a set of values (<L, +L, <B, +B, etc).
Now I defined a window, and two window functions like this:
ts_w_lp = Window.partitionBy(df_lp['pid']).orderBy(df_lp['ts'].asc())
first_lp = fns.min(df_lp['ts']).over(ts_w_lp)
last_lp = fns.max(df_lp['ts']).over(ts_w_lp)
I want to find the first and last timestamps for a given pid, and action. So I do this:
df_lp_pB = df_lp.filter(df_lp['action'] == '+B')\
.select('pid', first_lp.alias('tsf'), last_lp.alias('tsl'))\
.distinct().sort('pid')
However the dataframe I get has extra rows where one of the window functions seem to have intermediate values.
df_lp_pB.filter('pid = "BAG26723881"').toPandas()
pid tsf tsl
0 BAG26723881 2017-04-11 15:10:35.674 2017-04-11 15:10:35.674
1 BAG26723881 2017-04-11 15:10:35.674 2017-04-11 15:10:35.736
If I do the same with Spark SQL, it works as expected.
df_lp.createOrReplaceTempView('scans_lp')
df_sql = spark.sql("SELECT pid , min(ts) AS tsf, max(ts) AS tsl FROM scans_lp \
WHERE action='+B' GROUP BY pid ORDER BY pid")
df_sql.filter('pid = "BAG26723881"').toPandas()
pid tsf tsl
0 BAG26723881 2017-04-11 15:10:35.674 2017-04-11 15:10:35.736
In fact, when I invert the sorting of the timestamp column to descending, the other window function has the issue!
ts_w_lp2 = Window.partitionBy(df_lp['pid']).orderBy(df_lp['ts'].desc())
first_lp2 = fns.min(df_lp['ts']).over(ts_w_lp2)
last_lp2 = fns.max(df_lp['ts']).over(ts_w_lp2)
df_lp_pB2 = df_lp.filter(df_lp['action'] == '+B')\
.select('pid', first_lp2.alias('tsf'), last_lp2.alias('tsl'))\
.distinct().sort('pid')
df_lp_pB2.filter('pid = "BAG26723881"').toPandas()
pid tsf tsl
0 BAG26723881 2017-04-11 15:10:35.736 2017-04-11 15:10:35.736
1 BAG26723881 2017-04-11 15:10:35.674 2017-04-11 15:10:35.736
If I investigate further, I see all my pids are considered distinct, even for rows they are not! See this:
df_lp.filter('action = "+B"').select('pid').distinct().count()
6382
df_sql.count()
6382
df_lp.filter('action = "+B"').select('pid').count()
120303
df_lp_pB.count()
120303
What is going on? Did I misunderstand what Window.partitionBy() is supposed to do?