Spark自定义Aggregator疑问：ARRAY<BOOLEAN>转ARRAY<LONG>编码问题

核心解决方案：使用集合类型Encoder #

Spark的Encoders类提供了collection()方法，专门用于为Java集合类型生成对应的Encoder。针对你的List<Long>类型，直接用该方法即可完成bufferEncoder和outputEncoder的实现，无需手动构建序列化逻辑。

修改后的Encoder方法代码：

@Override
public Encoder<List<Long>> bufferEncoder() {
    return Encoders.collection(List.class, Encoders.LONG());
}

@Override
public Encoder<List<Long>> outputEncoder() {
    return Encoders.collection(List.class, Encoders.LONG());
}

代码优化建议 #

1. 处理数组内的null元素：当前reduce方法中，如果输入的List<Boolean>包含null元素，调用a.get(i)会抛出NullPointerException，建议修改判断逻辑：

if (a.get(i) != null && a.get(i)) {
    b.set(i, b.get(i) + 1);
}

2. 统一数组长度处理逻辑：在reduce方法中，当buffer的长度大于输入数组长度时，当前逻辑会直接遍历输入数组的长度，这部分行为合理，但可以添加注释说明：忽略buffer中超出输入数组长度的位置，保持原有计数。

3. 注册与使用示例：完成Aggregator实现后，需要将其注册为UDAF才能在DataFrame中使用：

// 实例化自定义Aggregator
ElementWiseAgg elementWiseAgg = new ElementWiseAgg();
// 注册为UDAF
UDAF<List<Long>> elementWiseSum = functions.udaf(elementWiseAgg, Encoders.collection(List.class, Encoders.LONG()));

// 在分组聚合中使用
df.groupBy("group_column")
  .agg(elementWiseSum.apply(col("boolean_array_column")).alias("element_wise_true_counts"))
  .show();

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