在按主鍵對記錄進行分組后，我正在使用聚合器在 DataFrame 上應用一些自定義合并：
case class Player(
pk: String,
ts: String,
first_name: String,
date_of_birth: String
)
case class PlayerProcessed(
var ts: String,
var first_name: String,
var date_of_birth: String
)
// Cutomer Aggregator -This just for the example, actual one is more complex
object BatchDedupe extends Aggregator[Player, PlayerProcessed, PlayerProcessed] {
def zero: PlayerProcessed = PlayerProcessed("0", null, null)
def reduce(bf: PlayerProcessed, in : Player): PlayerProcessed = {
bf.ts = in.ts
bf.first_name = in.first_name
bf.date_of_birth = in.date_of_birth
bf
}
def merge(bf1: PlayerProcessed, bf2: PlayerProcessed): PlayerProcessed = {
bf1.ts = bf2.ts
bf1.first_name = bf2.first_name
bf1.date_of_birth = bf2.date_of_birth
bf1
}
def finish(reduction: PlayerProcessed): PlayerProcessed = reduction
def bufferEncoder: Encoder[PlayerProcessed] = Encoders.product
def outputEncoder: Encoder[PlayerProcessed] = Encoders.product
}
val ply1 = Player("12121212121212", "10000001", "Rogger", "1980-01-02")
val ply2 = Player("12121212121212", "10000002", "Rogg", null)
val ply3 = Player("12121212121212", "10000004", null, "1985-01-02")
val ply4 = Player("12121212121212", "10000003", "Roggelio", "1982-01-02")
val seq_users = sc.parallelize(Seq(ply1, ply2, ply3, ply4)).toDF.as[Player]
val grouped = seq_users.groupByKey(_.pk)
val non_sorted = grouped.agg(BatchDedupe.toColumn.name("deduped"))
non_sorted.show(false)
這將回傳：
-------------- --------------------------------
|key |deduped |
-------------- --------------------------------
|12121212121212|{10000003, Roggelio, 1982-01-02}|
-------------- --------------------------------
現在，我想ts在匯總記錄之前對記錄進行排序。從這里我了解到，.sortBy("ts")不保證之后的訂單.groupByKey(_.pk)。所以我試圖在和.sortBy之間應用.groupByKey.agg
的輸出.groupByKey(_.pk)是 a KeyValueGroupedDataset[String,Player]，是第二個元素 a Iterator。因此，為了應用一些排序邏輯，我將其轉換為Seq：
val sorted = grouped.mapGroups{case(k, iter) => (k, iter.toSeq.sortBy(_.ts))}.agg(BatchDedupe.toColumn.name("deduped"))
sorted.show(false)
但是，.mapGroups添加排序邏輯后的輸出是Dataset[(String, Seq[Player])]. 因此，當我嘗試在.agg其上呼叫該函式時，出現以下例外：
Caused by: ClassCastException: org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema cannot be cast to $line050e0d37885948cd91f7f7dd9e3b4da9311.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$Player
我怎樣才能將 my 的輸出轉換回.mapGroups(...)a KeyValueGroupedDataset[String,Player]？
我試圖按如下方式轉換回迭代器：
val sorted = grouped.mapGroups{case(k, iter) => (k, iter.toSeq.sortBy(_.ts).toIterator)}.agg(BatchDedupe.toColumn.name("deduped"))
但是這種方法產生了以下例外：
UnsupportedOperationException: No Encoder found for Iterator[Player]
- field (class: "scala.collection.Iterator", name: "_2")
- root class: "scala.Tuple2"
我還能如何在.groupByKey和.agg方法之間添加排序邏輯？
uj5u.com熱心網友回復：
基于上面的討論， 的目的是通過忽略值Aggregator來獲取最新的欄位值。Playertsnull
這可以很容易地使用單獨聚合所有欄位來實作max_by。這樣就不需要自定義Aggregator或可變聚合緩沖區。
import org.apache.spark.sql.functions._
val players: Dataset[Player] = ...
// aggregate all columns except the key individually by ts
// NULLs will be ignored (SQL standard)
val aggColumns = players.columns
.filterNot(_ == "pk")
.map(colName => expr(s"max_by($colName, if(isNotNull($colName), ts, null))").as(colName))
val aggregatedPlayers = players
.groupBy(col("pk"))
.agg(aggColumns.head, aggColumns.tail: _*)
.as[Player]
在最新版本的 Spark 上，您還可以使用內置max_by運算式：
import org.apache.spark.sql.functions._
val players: Dataset[Player] = ...
// aggregate all columns except the key individually by ts
// NULLs will be ignored (SQL standard)
val aggColumns = players.columns
.filterNot(_ == "pk")
.map(colName => max_by(col(colName), when(col(colName).isNotNull, col("ts"))).as(colName))
val aggregatedPlayers = players
.groupBy(col("pk"))
.agg(aggColumns.head, aggColumns.tail: _*)
.as[Player]
轉載請註明出處，本文鏈接：https://www.uj5u.com/yidong/496929.html
上一篇：包內的隱式決議