如何通过Spark DataFrame基于ObjectId列表过滤MongoDB集合？

问题原因 #

你遇到的错误是因为com.mongodb.spark.sql.fieldTypes.ObjectId是Mongo Spark Connector的自定义类型，Spark的isin函数不支持直接传入该类型的对象列表，导致无法生成合法的过滤表达式。

解决方案 #

根据你的场景，推荐以下几种可行方案，优先选择无需将数据拉取到Driver端的方式（避免内存溢出风险）：

方案1：字符串匹配（修改agentIds提取方式）

既然Users集合中userId.oid是字符串类型，直接将Agents的agentId作为字符串列表，匹配该字段即可：

// 提取字符串类型的agentId并去重，生成字符串列表
val agentIdsList = agentIdsDF.select("agentId").distinct().as[String].collect().toList
// 过滤Users集合，匹配userId.oid字段
val usersDF = usersCollDF.filter(col("userId.oid").isin(agentIdsList:_*))

方案2：Spark SQL子查询（适合大数据量，避免Driver内存压力）

无需将agentIds拉取到Driver端，直接用子查询关联：

// 创建agentIds的临时视图
agentIdsDF.select("agentId").distinct().createOrReplaceTempView("distinct_agent_ids")
// 通过Spark SQL过滤匹配的用户文档
val usersDF = spark.sql("""
    SELECT * FROM usersCollDF
    WHERE userId.oid IN (SELECT agentId FROM distinct_agent_ids)
""")

方案3：DataFrame Join（最推荐的大数据场景方案）

用Join替代过滤，性能更优且避免Driver端数据拉取：

// 准备去重后的agentId DF，重命名字段以便关联
val distinctAgentIdsDF = agentIdsDF.select(col("agentId").alias("oid")).distinct()
// 内关联获取匹配的用户文档
val usersDF = usersCollDF.join(
    distinctAgentIdsDF, 
    usersCollDF("userId.oid") === distinctAgentIdsDF("oid"), 
    "inner"
)
// 可选：移除关联时新增的oid字段
val finalUsersDF = usersDF.drop("oid")

注意事项 #

• 避免使用collect()拉取大量数据到Driver端，否则可能引发内存溢出问题，优先选择子查询或Join方案。
• 确认Users集合的过滤字段是userId.oid（你原代码中写的col("user")可能是笔误）。

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