Spark：基于3天滚动时间窗口按组查询出现频次最高的ID

嘿，这个方案完全可行！在Spark 2.3及以下版本里，确实没法直接用高阶窗口函数一键搞定这个需求，但你提到的两个思路——借助collect_list结合自定义UDF，或者编写Scala自定义聚合函数（UDAF），都是实打实能跑通的解决方案。下面我给你拆解下两种方法的具体实现和注意事项：

这个方法实现起来更简单，适合窗口内数据量不算特别大的场景：

1. 定义滚动时间窗口
   首先要针对device_id分区，按read_date排序，设置3天的滚动范围。如果read_date是DateType，可以直接用天数来指定窗口范围，非常直观：

   import org.apache.spark.sql.expressions.Window
   import org.apache.spark.sql.functions._
   
   val rollingWindow = Window
     .partitionBy("device_id")
     .orderBy("read_date")
     .rangeBetween(-2, 0) // 包含当前日期及前2天，共3天
   

   如果read_date是字符串类型，记得先转成日期类型：to_date(col("read_date"))。

2. 收集窗口内的目标ID列表
   用collect_list把每个滚动窗口内的目标ID（假设字段名为target_id）收集成一个列表：

   val withIdListDF = df.withColumn("window_id_list", collect_list("target_id").over(rollingWindow))
   

3. 编写UDF找出频次最高的ID
   写一个Scala UDF，接收ID列表，统计每个ID的出现次数，然后返回频次最高的那个（如果有多个并列最高，可以根据业务需求调整逻辑，比如返回第一个、用逗号拼接所有等）：

   import scala.collection.mutable
   
   val getMostFrequentId = udf((idList: Seq[String]) => {
     val countMap = mutable.HashMap[String, Int]().withDefaultValue(0)
     idList.foreach(id => countMap(id) += 1)
     // 按频次降序排序，取第一个ID；若列表为空返回null
     countMap.toList.sortBy(-_._2).headOption.map(_._1).getOrElse(null)
   })
   

4. 应用UDF得到结果

   val resultDF = withIdListDF.withColumn("most_freq_id", getMostFrequentId(col("window_id_list")))
   

优缺点：

• ✅ 优点：代码简洁，开发成本低，不需要理解复杂的UDAF逻辑
• ❌ 缺点：如果滚动窗口内数据量极大，collect_list会把所有ID加载到Executor内存中，容易触发OOM，只适合中小规模的窗口数据

如果你的窗口数据量很大，UDAF是更高效的选择——它会增量统计ID频次，不需要存储整个窗口的ID列表：

1. 实现弱类型UDAF（Spark 2.3仅支持弱类型）
   继承UserDefinedAggregateFunction，实现初始化、更新、合并、求值四个核心方法：

   import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
   import org.apache.spark.sql.types._
   import scala.collection.mutable
   
   class MostFrequentIdUDAF extends UserDefinedAggregateFunction {
     // 输入类型：目标ID的类型（这里假设是String）
     override def inputSchema: StructType = StructType(StructField("target_id", StringType) :: Nil)
   
     // 缓冲区类型：存储ID计数Map、当前最高频次、对应ID
     override def bufferSchema: StructType = StructType(
       StructField("count_map", MapType(StringType, IntegerType)) ::
       StructField("max_count", IntegerType) ::
       StructField("most_freq_id", StringType) :: Nil
     )
   
     // 输出类型：频次最高的ID
     override def dataType: DataType = StringType
   
     override def deterministic: Boolean = true
   
     // 初始化缓冲区
     override def initialize(buffer: MutableAggregationBuffer): Unit = {
       buffer(0) = mutable.HashMap[String, Int]().toMap
       buffer(1) = 0
       buffer(2) = null
     }
   
     // 新增一条数据时更新缓冲区
     override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
       if (!input.isNullAt(0)) {
         val id = input.getString(0)
         val countMap = buffer.getAs[mutable.HashMap[String, Int]](0)
         countMap(id) += 1
         val currentCount = countMap(id)
   
         // 更新最高频次及对应ID
         if (currentCount > buffer.getInt(1)) {
           buffer(1) = currentCount
           buffer(2) = id
         } else if (currentCount == buffer.getInt(1)) {
           // 处理并列情况：比如保留第一个，或拼接所有ID
           // buffer(2) = s"${buffer.getString(2)},$id"
         }
         buffer(0) = countMap.toMap
       }
     }
   
     // 合并两个缓冲区（Shuffle阶段用）
     override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
       val map1 = buffer1.getAs[mutable.HashMap[String, Int]](0)
       val map2 = buffer2.getAs[Map[String, Int]](0)
       // 合并两个计数Map
       map2.foreach { case (id, count) => map1(id) += count }
   
       // 重新计算最高频次和对应ID
       val maxEntry = map1.toList.sortBy(-_._2).headOption
       maxEntry match {
         case Some((id, count)) =>
           buffer1(1) = count
           buffer1(2) = id
         case None =>
           buffer1(1) = 0
           buffer1(2) = null
       }
       buffer1(0) = map1.toMap
     }
   
     // 返回最终结果
     override def evaluate(buffer: Row): Any = buffer.getString(2)
   }
   

2. 注册并应用UDAF

   val mostFrequentIdUDAF = new MostFrequentIdUDAF()
   val resultDF = df.withColumn("most_freq_id", mostFrequentIdUDAF(col("target_id")).over(rollingWindow))
   

优缺点：

• ✅ 优点：内存效率高，增量统计避免加载全量窗口数据，适合大窗口场景
• ❌ 缺点：代码复杂度高，需要处理缓冲区的各种边界情况（比如空值、合并逻辑）

• 窗口范围的准确性：如果read_date包含时间（比如TimestampType），记得用时间戳的秒数来定义rangeBetween，比如rangeBetween(-2*86400, 0)（2天的秒数），避免把同一天的不同时间点排除在外。
• 并列最高的处理：两种方法都可以根据业务需求调整并列情况的逻辑，比如返回所有频次最高的ID，或者按ID排序取第一个。

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