How to calculate unique session id per user per minute in a click stream dataset using Spark-SQL?

Imagine that we have a clickstream dataset having millions of rows. And we want to calculate unique session-id per user per minute. Sample Dataset :

+------+-------------------+
|userId|clicktime          |
+------+-------------------+
|1039  |2009-04-21 13:17:50|
|1039  |2009-04-21 13:17:59|
|1039  |2009-04-21 13:19:59|
|1038  |2009-05-21 13:17:50|
|1037  |2009-05-21 13:17:00|
|1037  |2009-05-21 13:17:50|
|1037  |2009-05-21 13:17:59|
|1037  |2009-05-21 13:19:59|
|1038  |2009-05-21 13:19:59|
|1039  |2009-04-21 13:20:50|
+------+-------------------+

I have written a code in Spark-Scala to solve this but it is not an optimal solution for dataset having millions of rows. I want a better solution than what I have implemented. Below is the source code of my implementation:

val dfWithLag = rawData
      .withColumn("lag", lag(col("clicktime"), 1)
        .over(Window.partitionBy("userId") orderBy ("clicktime")).cast("timestamp"))
      .withColumn("lag_diff", unix_timestamp($"clicktime") - unix_timestamp($"lag"))
      .withColumn("lag_diff", when(col("lag_diff").isNull, 0).otherwise(col("lag_diff")))
      .orderBy("userId", "clicktime")


    val finalDf = dfWithLag.repartition(col("userId")).mapPartitions(partition => {
      var sessionId = scala.util.Random
      var currentSessionId = sessionId.nextInt().toInt
      val newPartition = partition
        .map(record => {
          ClickStream(record.getInt(0),record.getTimestamp(1), record.getTimestamp(2),
            record.getLong(3), {
              val timeDiff = record.getLong(3)
              if (timeDiff > 60) {
                currentSessionId = sessionId.nextInt.toInt
                currentSessionId
              }
              else if (timeDiff == 0) currentSessionId
              else currentSessionId
            }
          )
        }).toList
      newPartition.iterator
    })
    (Encoders.product[ClickStream])

    rawData.show(false)
    finalDf.drop("lag").drop("lagDiff").show(false)

Output of the code:

+------+-------------------+-----------+
|userId|clickTime          |sessionId  |
+------+-------------------+-----------+
|1037  |2009-05-21 13:17:00|1049786501 |
|1037  |2009-05-21 13:17:50|1049786501 |
|1037  |2009-05-21 13:17:59|1049786501 |
|1037  |2009-05-21 13:19:59|-1649908351|
|1039  |2009-04-21 13:17:50|-1794290301|
|1039  |2009-04-21 13:17:59|-1794290301|
|1039  |2009-04-21 13:19:59|668855070  |
|1039  |2009-04-21 13:20:50|668855070  |
|1038  |2009-05-21 13:17:50|1149727960 |
|1038  |2009-05-21 13:19:59|-95969967  |
+------+-------------------+-----------+