val avgfuel = events
    .groupBy(window($"enqueuedTime", "30 seconds"), $"weatherCondition")
    .agg(avg($"fuelEfficiencyPercentage") as "avg_fuel")

val joined = events.join(avgfuel, Seq("weatherCondition"))
    .filter($"fuelEfficiencyPercentage" > $"avg_fuel")

val streamingQuery1 = joined.writeStream
    .outputMode("append").
    .trigger(Trigger.ProcessingTime("10 seconds")).
    .option("checkpointLocation", checkpointLocation).
    .format("json").option("path", containerOutputLocation).start()