Pyspark: Average of timestamp

IIUC, just convert the working_hour to unix_timestamp and then take the average:

from pyspark.sql.functions import expr

df.withColumn('working_hour', expr('unix_timestamp(concat("2019-01-01", right(working_hour,9)))')) \
  .groupby('user_name') \
  .agg(expr("from_unixtime(avg(working_hour),'HH:mm:ss') as avg_hour")) \
  .show()
+---------+--------+                                                            
|user_name|avg_hour|
+---------+--------+
|      DEF|08:50:00|
|      XYZ|08:10:01|
|      ABC|09:20:01|
+---------+--------+

Or use DSL functions:

from pyspark.sql import functions as F

df.withColumn('working_hour', F.unix_timestamp(F.date_format('working_hour', '2019-01-01 HH:mm:ss'))) \
   .groupby('user_name') \
   .agg(F.from_unixtime(F.avg('working_hour'),'HH:mm:ss').alias('avg_hour')) \
   .show()

Some explanation:

(1) we use expr function to wrap up an SQL expression and return the result into a column. this is a convenient way to access SparkSQL's builtin functions, see doc.

(2) use concat("2019-01-01", right(working_hour,9)) to convert all the working_hour into the same date so that date difference won't influence the calculation of average. (here we concatenate the string "2019-01-01" with the last 9 characters of working_hour). (notice that the SparkSQL right function will cast the first argument into string internally). see also date_format function to handle the same.

(3) then use unix_timestamp to convert the above result into unix timestamp (bigint)

(4) groupby(user_name) and take the avg of the result from (3), and then convert it back to the time format using from_unixtime.

I tried approaching this as below-

1. Read the input

 val spark = sqlContext.sparkSession
    val implicits = spark.implicits
    import implicits._
    val schema = StructType(
      Array(StructField("user_name", DataTypes.StringType),
        StructField("working_hour", DataTypes.TimestampType))
      )
    val data1 =
      """
        |  ABC        |2019-10-21 09:00:02
        |  XYZ        |2019-10-21 09:15:01
        |  DEF        |2019-10-21 08:55:00
        |  ABC        |2019-10-22 09:40:00
        |  XYZ        |2019-10-22 07:05:01
        |  DEF        |2019-10-22 08:45:00
      """.stripMargin

    val df1 = spark.read
      .schema(schema)
      .option("sep", "|")
      .csv(data1.split(System.lineSeparator()).map(_.replaceAll("""^\s*|\s$""", "")).toSeq.toDS())
    df1.show(false)
    df1.printSchema()

Result-

+-----------+-------------------+
|user_name  |working_hour       |
+-----------+-------------------+
|ABC        |2019-10-21 09:00:02|
|XYZ        |2019-10-21 09:15:01|
|DEF        |2019-10-21 08:55:00|
|ABC        |2019-10-22 09:40:00|
|XYZ        |2019-10-22 07:05:01|
|DEF        |2019-10-22 08:45:00|
+-----------+-------------------+

root
 |-- user_name: string (nullable = true)
 |-- working_hour: timestamp (nullable = true)

2. Find the average in seconds and convert it to HH:MM:SS

val processedDf = df1.
      withColumn("diff", unix_timestamp(col("working_hour")) - unix_timestamp(to_date(col("working_hour"))))
      .groupBy("user_name")
      .agg(avg("diff").cast(LongType).as("avg"))


   val HHMMSS =  udf((seconds: Long) => {
      val df = new java.text.SimpleDateFormat("HH:mm:ss")
      df.setTimeZone(java.util.TimeZone.getTimeZone("GMT"))
      df.format(seconds * 1000)
    })
    processedDf.withColumn("hhmmss", HHMMSS(col("avg")))
      .show(false)

Result-

+-----------+-----+--------+
|user_name  |avg  |hhmmss    |
+-----------+-----+--------+
|ABC        |33601|09:20:01|
|DEF        |31800|08:50:00|
|XYZ        |29401|08:10:01|
+-----------+-----+--------+