如何使用scala通过spark中的一个或多个字符串参数传递selectExpr中的列名？

Question

我在 Spark Streaming 中使用 CDC Merge 脚本。我希望通过参数传递 selectExpr 中的列值，因为每个 table 的列名都会改变。当我通过字符串变量传递列和结构字段时，出现错误 ==> 不匹配的输入 ',' 期望

下面是我要参数化的代码。

var filteredMicroBatchDF=microBatchOutputDF
.selectExpr("col1","col2","struct(offset,KAFKA_TS) as otherCols" )
.groupBy("col1","col2").agg(max("otherCols").as("latest"))
.selectExpr("col1","col2","latest.*")

参考我正在尝试模拟的脚本：- https://docs.databricks.com/_static/notebooks/merge-in-cdc.html

我试过如下所示，通过在变量中传递列名，然后从这些变量中读取 selectExpr：-

val keyCols = "col1","col2"
val structCols = "struct(offset,KAFKA_TS) as otherCols" 

var filteredMicroBatchDF=microBatchOutputDF
.selectExpr(keyCols,structCols )
.groupBy(keyCols).agg(max("otherCols").as("latest"))
.selectExpr(keyCols,"latest.*")

当我运行脚本时，它给我错误 org.apache.spark.sql.streaming.StreamingQueryException: mismatched input ',' expecting <<EOF>>

编辑

这是我在 Luis Miguel 的评论后尝试过的，效果很好：-

import org.apache.spark.sql.{DataFrame, functions => sqlfun}

def foo(microBatchOutputDF: DataFrame)
       (keyCols: Seq[String], structCols: Seq[String]): DataFrame =
  microBatchOutputDF
    .selectExpr((keyCols ++ structCols) : _*)
    .groupBy(keyCols.head, keyCols.tail : _*).agg(sqlfun.max("otherCols").as("latest"))
    .selectExpr((keyCols :+ "latest.*") : _*)

var keyColumns = Seq("COL1","COL2")
var structColumns = "offset,Kafka_TS"

foo(microBatchOutputDF)(keyCols = Seq(keyColumns:_*), structColumns = Seq("struct("+structColumns+") as otherCols"))

注：下面会报错

foo(microBatchOutputDF)(keyCols = Seq(keyColumns), structColumns = Seq("struct("+structColumns+") as otherCols"))

关于上面的工作代码的事情是，这里的 keyColumns 是硬编码的。因此，我尝试（首先）从参数文件读取和（其次）从导致错误的小部件读取，我在这里寻找建议和建议：-

第一种方法

def loadProperties(url: String):Properties = {
    val properties: Properties = new Properties()
    if (url != null) {
      val source = Source.fromURL(url)
      properties.load(source.bufferedReader())
    }
  return properties
}
var tableProp: Properties = new Properties()
tableProp = loadProperties("dbfs:/Configs/Databricks/Properties/table/Table.properties") 
var keyColumns = Seq(tableProp.getProperty("keyCols"))
var structColumns = tableProp.getProperty("structCols")

keyCols 和 StructCols 在参数文件中定义为：-

keyCols = Col1, Col2 （我也尝试将它们分配为 "Col1"，"Col2"）
StructCols = offset,Kafka_TS

最后，

foo(microBatchOutputDF)(keyCols = Seq(keyColumns:_*), structColumns = Seq("struct("+structColumns+") as otherCols"))

代码抛出指向第一个逗号的错误（就好像它将列字段作为单个参数）：
mismatched input ',' expecting <EOF>
== SQL ==
"COL1","COL2""
-----^^^

如果我只传递 keyCols 属性中的一列，代码工作正常。
例如。 keyCols = Col1

第二种方法
在这里，我尝试从小部件读取关键列，但再次出现相同的错误。

dbutils.widgets.text("prmKeyCols", "","") 
val prmKeyCols = dbutils.widgets.get("prmKeyCols") 
var keyColumns = Seq(prmKeyCols)

小部件传入如下
"Col1","Col2"

最后，

foo(microBatchOutputDF)(keyCols = Seq(keyColumns:_*), structColumns = Seq("struct("+structColumns+") as otherCols"))

这也给出了同样的错误。

Answer 1

像这样的东西应该可以工作：

import org.apache.spark.sql.{DataFrame, functions => sqlfun}

def foo(microBatchOutputDF: DataFrame)
       (keyCols: Seq[String], structCols: Seq[String]): DataFrame =
  microBatchOutputDF
    .selectExpr((keyCols ++ structCols) : _*)
    .groupBy(keyCols.head, keyCols.tail : _*).agg(sqlfun.max("otherCols").as("latest"))
    .selectExpr((keyCols :+ "latest.*") : _*)

你可以像这样使用：

foo(microBatchOutputDF)(keyCols = Seq("col1", "col2"), structCols = Seq("struct(offset,KAFKA_TS) as otherCols"))

如何使用scala通过spark中的一个或多个字符串参数传递selectExpr中的列名？

How to pass column names in selectExpr through one or more string parameters in spark using scala?

scala

apache-spark

spark-streaming

apache-spark-sql