python - Pyspark:时间戳的平均值

Question

我有一个数据框，看起来像:-

+--------------------+-------------------+
|      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|
+--------------------+-------------------+

我想要每个用户的平均工作时间。我在不同日期有工作时间。我正在尝试使用 groupBy 用户。它是这样给出的。

df_avg = df_work_hours.groupBy('user_name').agg(sql_functions.avg('working_hour').alias('avg_hour'))

+--------------------+--------------------+
|      user_name     |            avg_hour|
+--------------------+--------------------+
|         ABC        |      1.5717579755E9|
|         XYZ        |1.5717276003333333E9|
|         DEF        |          1.571658E9|
+--------------------+--------------------+

但我想喜欢这个:-

+--------------------+------------------+
|      user_name     | avg(working_hour)|
+--------------------+------------------+
|         ABC        |          09:20:01|
|         XYZ        |          08:10:01|
|         DEF        |          08:50:00|
+--------------------+------------------+

Answer 1

IIUC，只需将working_hour转换为unix_timestamp，然后取平均值即可:

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|
+---------+--------+

或者使用 DSL 函数:

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()

一些解释:

(1) 我们使用 expr函数包装 SQL 表达式并将结果返回到列中。这是访问 SparkSQL 内置函数的便捷方法，请参阅 doc 。

(2) 使用 concat("2019-01-01", right(working_hour,9)) 将所有 working_hour 转换为同一日期，以便该日期差异不会影响平均值的计算。 (这里我们将字符串“2019-01-01”与working_hour的最后9个字符连接起来)。 (请注意，SparkSQL right 函数将在内部将第一个参数转换为字符串)。另请参阅date_format函数来处理相同的问题。

(3) 然后使用 unix_timestamp将上面的结果转换成unix时间戳(bigint)

(4) groupby(user_name) 并获取 (3) 结果的 avg，然后使用 from_unixtime 将其转换回时间格式.