From Lag function to updateStateByKey in pyspark streaming Dstream

Question

I have readstream data came from kafka topic , I changed to a dataframe (because I am using structured streamin pyspark)

This is my schema:

    root
 |-- DriverId: string (nullable = true)
 |-- time: timestamp (nullable = true)
 |-- Longitude: float (nullable = true)
 |-- Latitude: float (nullable = true)
 |-- SPEED: integer (nullable = true)
 |-- EngineSpeed: string (nullable = true)
 |-- MAF: string (nullable = true)
 |-- FuelType: string (nullable = true)

Harsh acceleration is an increase in speed of 6.14 miles per hour in one second.

So I calculate this on Batch (I read csv file an apply this):

acc_max=9.87926
dec_max=10.9412
#w = Window.partitionBy('DriverId').orderBy('time')
df = df.withColumn("prev_value",F.lag(df.SPEED).over(w))
df = df.withColumn("diff_speed", F.when(F.isnull(df.SPEED - df.prev_value), 0)
                              .otherwise(df.SPEED - df.prev_value))
df = df.withColumn(
    'HarshAcceleration',
    F.when((F.col("diff_speed") > acc_max),1)\
    .otherwise(0))
df = df.withColumn(
    'HarshBraking',
    F.when((F.col("diff_speed") < - dec_max),1)\
    .otherwise(0))
df=(df.groupBy("DriverId").sum("harshAcceleration").alias("harshAcceleration")).show()

and it shows this result:

+--------+----------------------+
|DriverId|sum(harshAcceleration)|
+--------+----------------------+
|       3|                     4|
|       0|                     1|
|       1|                     1|
|       2|                     0|
+--------+----------------------+

Now I want do this but on continuous streaming , I can't use lag function multiple errors I have found!

windowSpec  = Window.partitionBy("DriverId").orderBy(df.time)
acc_max=9.87926
df=df.withColumn("prev_value",F.lag(df.SPEED).over(windowSpec))
df=df.withColumn("diff_speed", F.when(F.isnull(df.SPEED - df.prev_value), 0)\
    .otherwise(df.SPEED - df.prev_value))
df=df.withColumn('HarshAcceleration',F.when((F.col("diff_speed") > acc_max),1) \
  .otherwise(0))
df=df.groupBy(window(df.time,"3 seconds"),df.DriverId).sum("HarshAcceleration").orderBy('window')
    # Start running the query that prints the windowed word counts to the console
query=df\
    .select("DriverId","sum(HarshAcceleration)")\
    .writeStream\
    .format("console") \
    .outputMode("complete")\
    .start()

but i get this error:

AnalysisException: 'Non-time-based windows are not supported on streaming DataFrames/Datasets;; Window [lag(SPEED#41, 1, null) windowspecdefinition(DriverId#25, time#68 ASC NULLS FIRST, specifiedwindowframe(RowFrame, -1, -1)) AS prev_value#78], [DriverId#25], [time#68 ASC NULLS FIRST] +- Project [DriverId#25, time#68, Longitude#59, Latitude#50, SPEED#41, EngineSpeed#30, MAF#31, FuelType#32] +- Project [DriverId#25, cast(time#26 as timestamp) AS time#68, Longitude#59, Latitude#50, SPEED#41, EngineSpeed#30, MAF#31, FuelType#32]

can you please suggest for a solution with pysparkstreaming rdd with updatebykey

From Lag function to updateStateByKey in pyspark streaming Dstream

0 Answers