Spark中字符串过滤条件转换为Column类型的实现方法咨询

要将字符串格式的过滤条件转为Spark的Column类型，直接用Spark SQL提供的expr函数即可，该函数可以将符合SQL语法的字符串表达式直接解析为Column对象。

修正步骤 #

• 首先引入依赖：import org.apache.spark.sql.functions.expr
• 调整传入的filcond字符串格式：无需使用Scala DSL的col("xxx") === "xxx"写法，直接写标准SQL表达式即可，比如过滤custId为1的条件直接写为val filcond = "custId = '1'"
• 修改myFilter函数的custom分支，返回expr(filcond)

修正后代码 #

import org.apache.spark.sql.functions.expr

// 注意补充dt、dB两个参数的入参声明
def myFilter(mode: String, filcond : String, dt: String, dB: Int): Column = {
  if (mode == "daily") {
    $"update_date" >= date_sub(to_date(date_format(lit(dt), "yyyy-MM-dd")), dB) and $"update_date" <= to_date(date_format(lit(dt), "yyyy-MM-dd"))
  } else {
    expr(filcond)
  }
}

// 调用示例
val mode = "custom"
val filcond = "custId = '1'" // 直接写SQL格式的过滤条件
val myDf = ss.read
  .parquet("/data")
  .select (
    $"Id",
    $"Url",
    $"Type",
    $"custId",
    to_date($"updateTimestamp","yyyy-MM-dd").as("update_date")
  )
  .filter(myFilter(mode, filcond, "2024-01-01", 7))

注意事项 #

• 传入的过滤条件字符串必须符合Spark SQL表达式语法，否则运行时会抛出表达式解析异常
• 如果过滤条件包含动态传入的参数，建议字符串拼接时做好特殊字符转义，避免语法错误或者注入风险
• 如果一定要用Scala DSL风格的字符串作为输入，需要用到Scala反射机制动态解析代码，实现复杂度高且存在安全风险，远不如直接用SQL表达式+expr的方案高效可靠

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