Витягне значення стовпців в іскрових кадрах даних - scala, apache-spark, apache-spark-sql, spark-dataframe, spark-streaming

У мене є вимога, де мені потрібно відфільтруватирядки з іскрового кадру даних, де значення певного стовпця (скажімо, "ціна") має співпадати з значеннями, наявними на карті scala. Ключ scala map - це значення іншого стовпця (скажімо, "id"). Мій кадр даних містить два стовпці: ідентифікатор та ціна. Мені потрібно відфільтрувати всі стовпці, де ціна не відповідає ціні, зазначеній на карті Scala.

Мій код нагадує це:

object obj1{
// This method returns value price for items as per their id
getPrice(id:String):String {
//lookup in a map and return the price
}
}

object Main{
val validIds = Seq[String]("1","2","3","4")
val filteredDf = baseDataframe.where(baseDataframe("id").in(validIDs.map(lit(_)): _*) &&
baseDataframe("price") === (obj1.getPrice(baseDataframe("id").toString())))

// But this line send string "id" to obj1.getPrice() function
// rather than value of id column
}
}

Я не можу передавати значення id стовпців, щоб функціонувати obj1.getPrice (). Будь-яка пропозиція, як це досягти?

Дякую,

Відповіді:

Ви можете написати udf, щоб зробити це:

val checkPrice(id: String, price: String) = validIds.exists(_ == id) && obj1.getPrice(id) == price
val checkPriceUdf = udf(checkPrice)

baseDataFrame.where(checkPriceUdf($"id", $"price"))

Або інше рішення - конвертувати Map від id -> ціна до фрейму даних, а потім виконати внутрішнє об'єднання з baseDataFrame на id і price колони