Spark中如何通过其他DataFrame的列过滤DataFrame？报错排查

问题根源 #

代码里collect.toList.asInstanceOf[List[String]]存在错误：collect返回的是Array[Row]，转成List后每个元素是Row对象，强制类型转换并不会把Row转为String。当把这个列表传给isin时，Spark收到的是Row实例，而它无法处理这种类型的字面量，因此抛出The feature is not supported: literal for '[Agatha Christie]' of class org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema错误。

另外，authors列的字段值是带方括号的字符串（比如[Agatha Christie]），需要确保提取的字符串和目标列格式完全匹配。

解决方案 #

方法一：正确提取作者字符串列表（适合小数据集） #

先修正作者列表的提取逻辑，从Row中取出真实的字符串值：

// 提取高分作者的字符串列表
val userReviewsAuthorsManyRead = userReviewsWithBooksDetails
  .filter($"authors".isNotNull)
  .groupBy(col("authors"))
  .agg(
    count("review/score").alias("count"),
    avg("review/score").alias("avg_score")
  )
  .filter($"avg_score" >= 4)
  .sort($"count".desc, $"avg_score".desc)
  .limit(5)
  .select("authors")
  .collect()
  .map(row => row.getString(0)) // 从Row中取出authors字段的字符串值
  .toList

// 使用提取出的列表过滤目标DataFrame
val booksReadBestAuthors = userReviewsWithBooksDetails
  .filter(col("authors").isin(userReviewsAuthorsManyRead:_*))

方法二：使用DataFrame关联（推荐，适合大数据集） #

避免将数据collect到Driver节点，直接用Spark的关联操作更高效，符合分布式计算特性：

// 先构建高分作者的DataFrame
val topAuthorsDF = userReviewsWithBooksDetails
  .filter($"authors".isNotNull)
  .groupBy(col("authors"))
  .agg(
    count("review/score").alias("count"),
    avg("review/score").alias("avg_score")
  )
  .filter($"avg_score" >= 4)
  .sort($"count".desc, $"avg_score".desc)
  .limit(5)
  .select("authors")

// 通过内关联过滤目标DataFrame
val booksReadBestAuthors = userReviewsWithBooksDetails
  .join(topAuthorsDF, Seq("authors"), "inner")

可选：清洗authors列的格式 #

如果authors列的方括号是多余的（比如实际需要匹配的是Agatha Christie而非带方括号的字符串），可以先清洗字段：

import org.apache.spark.sql.functions.regexp_replace

// 去除authors字段的首尾方括号
val cleanedDF = userReviewsWithBooksDetails
  .withColumn("cleaned_authors", regexp_replace($"authors", "^\\[|\\]$", ""))

// 后续的分组、过滤、关联操作都基于cleaned_authors列即可

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