スパークSQLは、グループの中で最も一般的な項目を見つける

Question

次のように私には見えるの個別項目のグループを持っている場合は

case class Order(orderId: String, orderDetails: OrderDetail, destination: String) 
case class OrderDetail(date: Timestamp, recipient: String, item: String) 

grouped = ordersDF.groupby($"destination") 

何イムを探しているを見てみると、各宛先によって最も一般的な項目を見つけるための方法です集計に使用できるSQL関数、データのサブグループBysには何も表示されません。データをRDDに変換することはできますが、私の理解はそれがベストプラクティスではないということです。

私はあなたがGROUPBY /集約関数と窓関数の組み合わせを使用してこれを実現することができ

|Destination | mostCommon | 
---------------------------- 
|XYZ   |item x  | 

Answer 1

のようなものを見てみたいと思います。先とアイテムによって

+-------+--------------+-----------+ 
|orderId| orderDetails|destination| 
+-------+--------------+-----------+ 
|  1|[11,abc,item1]|  loc1| 
|  2|[12,abc,item2]|  loc1| 
|  3|[13,abc,item1]|  loc1| 
|  4|[14,abc,item1]|  loc2| 
|  5|[15,abc,item2]|  loc2| 
|  6|[11,abc,item2]|  loc2| 
|  7|[11,abc,item2]|  loc2| 
+-------+--------------+-----------+ 

まず、グループデータと、各項目の頻度を数える：

はのはordersDfされるように、これを考えてみましょう。

val dfWithCount = ordersDf 
.groupBy("destination","orderDetails.item") 
.agg(count("orderDetails.item").alias("itemCount")) 

我々は、場所ごとの最も一般的な項目を見つける目的地でのパーティションを聞かせITEMCOUNT個のカラムに最大アグリゲーションを適用したいと思いますので、集約されたデータフレームは、この

+-----------+-----+---------+ 
|destination| item|itemCount| 
+-----------+-----+---------+ 
|  loc1|item2|  1| 
|  loc2|item1|  1| 
|  loc2|item2|  3| 
|  loc1|item1|  2| 
+-----------+-----+---------+ 

のように見えます。

val maxWindowSpec = Window.partitionBy("destination") 
val maxColumn = max($"itemCount").over(maxWindowSpec) 
val dfWithMax = dfWithCount.withColumn("maxItemCount",maxColumn) 

は、得られたデータフレームは、宛先ごとの項目のitemCountsとMAXCOUNTの両方を持っている与えられた（宛先、アイテム）の組み合わせについてITEMCOUNT個が最大項目数ではありません

+-----------+-----+---------+------------+ 
|destination| item|itemCount|maxItemCount| 
+-----------+-----+---------+------------+ 
|  loc1|item2|  1|   2| 
|  loc1|item1|  2|   2| 
|  loc2|item1|  1|   3| 
|  loc2|item2|  3|   3| 
+-----------+-----+---------+------------+ 

最後に、我々は除外する行その目的地のために。

val result = dfWithMax 
.filter("maxItemCount - itemCount == 0") 
.drop("maxItemCount","itemCount") 

result.show() 

+-----------+-----+ 
|destination| item| 
+-----------+-----+ 
|  loc1|item1| 
|  loc2|item2| 
+-----------+-----+