如何在Scala Spark中按字典键值组合高效过滤DataFrame

要高效处理这类包含JSON数组字符串的DataFrame过滤需求，核心思路是先将JSON字符串解析为Spark可操作的数组结构，再通过高阶函数检查数组中是否存在符合条件的元素，避免不必要的数据展开（如explode）以减少性能开销。

步骤1：定义JSON数组的Schema #

为了高效解析JSON字符串，不要依赖自动推断Schema，手动定义Schema能大幅提升大数据量下的解析速度：

import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._

// 定义JSON数组中每个元素的结构
val jsonElementSchema = StructType(Seq(
  StructField("ID1", IntegerType),
  StructField("ID2", IntegerType),
  StructField("value", StringType)
))
// 外层是数组类型
val jsonArraySchema = ArrayType(jsonElementSchema)

步骤2：解析JSON字符串列为数组结构 #

将原始DataFrame中的JSON字符串列转换为Spark的数组结构列：

// 假设原始DataFrame的JSON列名为`json_str`
val parsedDF = originalDF.withColumn("parsed_json", from_json(col("json_str"), jsonArraySchema))

步骤3：过滤符合条件的行 #

使用Spark的exists高阶函数，检查数组中是否存在满足所有指定键值组合的元素。这种方式无需展开数组，性能更优：

// 目标键值组合：ID1=111、ID2=2、value=Z
val filteredDF = parsedDF.filter(
  exists(col("parsed_json"), element => 
    element("ID1") === lit(111) && 
    element("ID2") === lit(2) && 
    element("value") === lit("Z")
  )
).drop("parsed_json") // 不需要解析后的列可以删除

兼容Spark 2.x的写法 #

如果你的Spark版本是2.x（不支持exists高阶函数），可以用array_contains结合struct来实现：

// 构造目标结构
val targetStruct = struct(
  lit(111).as("ID1"),
  lit(2).as("ID2"),
  lit("Z").as("value")
)

val filteredDF = parsedDF.filter(array_contains(col("parsed_json"), targetStruct))
  .drop("parsed_json")

说明 #

• 对于缺少指定键的元素，element("键名")会返回null，与目标值比较结果为false，因此不会被误判为匹配，符合需求。
• 两种方法都避免了explode操作，不会产生额外的数据行，在大数据量场景下性能更优。

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