如何从Spark DataFrame的timestamp列按班次统计指定name的记录数

前置说明 #

所有逻辑均使用Spark原生内置函数实现，未使用UDF或toPandas()方法，最终输出为标准Spark DataFrame。

核心逻辑 #

利用Spark原生hour()函数提取timestamp字段的小时值，通过when()/otherwise()分支判断归属班次，分组聚合即可完成统计。小时判断逻辑完全匹配班次时间区间：比如11:59:59的小时值为11，归入12小时制的早班；15:59:59的小时值为15，归入8小时制的中班，无需额外处理秒级边界。

代码实现（PySpark） #

首先导入依赖函数：

from pyspark.sql import functions as F
# 假设原数据集DF名为source_df

1. 过滤目标用户数据 #

alice_df = source_df.filter(F.col("name") == "Alice")

2. 统计12小时制班次数量 #

shift_12h_result = alice_df.withColumn("shift",
    F.when(F.hour(F.col("timestamp")) < 12, F.lit("00:00-11:59:59"))
    .otherwise(F.lit("12:00-23:59:59"))
).groupBy("shift").agg(F.count("*").alias("record_count"))

3. 统计8小时制班次数量 #

shift_8h_result = alice_df.withColumn("shift",
    F.when(F.hour(F.col("timestamp")) < 8, F.lit("00:00-07:59:59"))
    .when(F.hour(F.col("timestamp")) < 16, F.lit("08:00-15:59:59"))
    .otherwise(F.lit("16:00-23:59:59"))
).groupBy("shift").agg(F.count("*").alias("record_count"))

4. 合并两种统计结果（可选） #

如果需要将两类班次的统计结果合并为同一个DataFrame，可增加班次类型标识后合并：

shift_12h_result = shift_12h_result.withColumn("shift_type", F.lit("12小时制"))
shift_8h_result = shift_8h_result.withColumn("shift_type", F.lit("8小时制"))
final_result = shift_12h_result.unionByName(shift_8h_result)

代码实现（Scala，可选） #

import org.apache.spark.sql.functions._

val aliceDf = sourceDf.filter(col("name") === "Alice")

// 12小时制统计
val shift12hResult = aliceDf.withColumn("shift",
    when(hour(col("timestamp")) < 12, lit("00:00-11:59:59"))
    .otherwise(lit("12:00-23:59:59"))
).groupBy("shift").agg(count("*").alias("record_count"))

// 8小时制统计
val shift8hResult = aliceDf.withColumn("shift",
    when(hour(col("timestamp")) < 8, lit("00:00-07:59:59"))
    .when(hour(col("timestamp")) < 16, lit("08:00-15:59:59"))
    .otherwise(lit("16:00-23:59:59"))
).groupBy("shift").agg(count("*").alias("record_count"))

内容的提问来源于stack exchange，提问作者Mario