Как передать имена столбцов в selectExpr через один или несколько строковых параметров в spark, используя scala? - PullRequest
Новая
       29

Как передать имена столбцов в selectExpr через один или несколько строковых параметров в spark, используя scala?

1 голос
/ 27 октября 2019

Я использую скрипт для CDC Merge в потоковой передаче. Я хочу передать значения столбца в selectExpr через параметр, поскольку имена столбцов для каждой таблицы изменились бы. Когда я пропускаю столбцы и поле структуры через строковую переменную, я получаю сообщение об ошибке ==> несоответствующий ввод ',' ожидающий

Ниже приведен фрагмент кода, который я пытаюсь параметризовать. 

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"))

Это также дает ту же ошибку.

1 Ответ

2 голосов
/ 27 октября 2019

Примерно так должно работать: 

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"))
...