如何基于Scala ENUM过滤Spark Dataset？

你遇到的错误是因为Spark在序列化Enumeration对象时，其内部的readResolve方法需要访问MODULE$字段，但由于Spark分布式执行环境的类加载机制限制，导致无法找到该字段。下面是几种可行的解决方法：

方法1：提前提取枚举值列表并复用 #

先将枚举中的有效部门字符串提取到普通变量中，再用该变量过滤Dataset，避免直接在filter闭包中引用Enumeration对象：

// 提前提取有效部门列表
val validDeptList = DeptType.toList
// 用普通列表过滤Dataset
val filteredDS = resDS.filter(rec => validDeptList.contains(rec.dept))
filteredDS.show()

方法2：使用Spark列表达式过滤 #

改用Spark的SQL风格列表达式过滤，这种方式无需将枚举对象序列化到分布式节点：

import org.apache.spark.sql.functions.col

val filteredDS = resDS.filter(col("dept").isin(DeptType.toList:_*))
filteredDS.show()

方法3：优化Enumeration实现（可选） #

如果需要保留枚举的contains方法，可以优化实现，避免依赖枚举对象的序列化：

object DeptType extends Enumeration {
  type DeptType = Value
  val CSE, ECE, EEE = Value

  // 提前缓存有效部门的字符串集合
  private val validDepts = values.map(_.toString).toSet
  def contains(s: String): Boolean = validDepts.contains(s)
}

// 使用时提前提取集合，避免闭包引用枚举对象
val validDepts = DeptType.validDepts
val filteredDS = resDS.filter(rec => validDepts.contains(rec.dept))

原理说明 #

Spark Dataset的强类型filter闭包会被序列化后发送到Worker节点执行，而Scala Enumeration对象的序列化/反序列化过程中，内部Val实例需要通过readResolve方法找到枚举单例，这在分布式环境中容易触发类加载异常。通过提前提取枚举值到普通集合，或者使用Spark列表达式，可绕过枚举对象的序列化问题。

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