Spark基于已有DataFrame值过滤Azure SQL表的报错解决

在Notebook中尝试从Azure SQL服务器导入两张大表，逻辑是先按日期过滤TableA，获取TableA中columnA的最小值，再用这个值过滤TableB以避免拉取海量数据，当前代码如下：

val date = "make_date(2023,06,01)"

val TableA= spark.read.synapsesql("sqlserver.dbo.table1")
.select("columnA"
,"columnB"
,"DateColumn")
.toDF()
.filter(expr("DateColumn>="+date))
.repartition(20)
.createOrReplaceTempView("TableA")


val min = spark.sql("""select min(columnA) as mincolA from TableA""".stripMargin)


val TableB = spark.read.synapsesql("sqlserver.dbo.tableB")
.select("columnA"
,"columnC")
.filter(expr("columnA >="+min))

执行时触发如下错误：

org.apache.spark.sql.catalyst.parser.ParseException:
===SQL===

columnA >= [min: bigint]

问题核心是min是Spark DataFrame对象，不是具体数值，无法直接嵌入SQL表达式做比较。需要先从DataFrame中提取出实际的最小值，转为Scala变量后再使用。

修正后的代码如下：

val date = "make_date(2023,06,01)"

val TableA= spark.read.synapsesql("sqlserver.dbo.table1")
.select("columnA", "columnB", "DateColumn")
.toDF()
.filter(expr(s"DateColumn >= $date"))
.repartition(20)
.createOrReplaceTempView("TableA")

// 提取DataFrame中的最小值，转为Long类型变量（若columnA是Int类型则改用getInt(0)）
val minColA = spark.sql("select min(columnA) as mincolA from TableA")
  .select("mincolA")
  .head()
  .getLong(0)

val TableB = spark.read.synapsesql("sqlserver.dbo.tableB")
.select("columnA", "columnC")
.filter(expr(s"columnA >= $minColA"))

关键说明 #

• 用head()获取DataFrame首行数据，再通过getLong(0)提取具体数值，需匹配columnA的实际数据类型
• 使用Scala字符串插值s""拼接变量，确保过滤条件中传入的是具体数值而非DataFrame对象
• 优化了select语句写法，合并为一行提升可读性

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