スパークでランクごとにデータセットを分離

Question

従業員名、残高、日付、従業員の名前を含むデータセットに別のランク番号があります。

df.show(); 
+------------+----------+-------+----+ 
| Employee|  date|balance|rank| 
+------------+----------+-------+----+ 
|  A  |2016-02-05| 2143| 1| 
|  A  |2016-07-05| 231| 2| 
|  A  |2016-08-05| 447| 3| 
|  A  |2017-10-05| 779| 4| 
|  A  |2018-03-05| 255| 5| 
|  A  |2018-05-05| 246| 6| 
|  A  |2018-08-05| 378| 7| 
|  A  |2018-11-05| 10635| 8| 
|  A  |2019-06-05|  49| 9| 
|  A  |2020-02-05|  0| 10| 
|  A  |2020-04-05| 244| 11| 
|  A  |2020-05-05|  0| 12| 
|  A  |2020-09-05| 424| 13| 
|  C  |2016-05-05| 1506| 1| 
|  C  |2017-06-05|  52| 2| 
|  C  |2017-09-05| 723| 3| 
|  C  |2017-11-05|  23| 4| 
+------------+----------+-------+----+ 

このデータセットはランクごとに分けなければなりません。ので、私の予想出力は

table1 
+------------+----------+-------+----+ 
| Employee|  date|balance|rank| 
+------------+----------+-------+----+ 
|  A  |2016-02-05| 2143| 1| 
|  A  |2016-07-05| 231| 2| 
|  A  |2016-08-05| 447| 3| 
|  A  |2017-10-05| 779| 4| 
|  A  |2018-03-05| 255| 5| 
|  A  |2018-05-05| 246| 6| 
|  A  |2018-08-05| 378| 7| 
|  A  |2018-11-05| 10635| 8| 
|  A  |2019-06-05|  49| 9| 
|  A  |2020-02-05|  0| 10| 
|  A  |2020-04-05| 244| 11| 
|  A  |2020-05-05|  0| 12| 
|  A  |2020-09-05| 424| 13| 
+------------+----------+-------+----+ 


table2 

+------------+----------+-------+----+ 
| Employee|  date|balance|rank| 
+------------+----------+-------+----+ 
|  C  |2016-05-05| 1506| 1| 
|  C  |2017-06-05|  52| 2| 
|  C  |2017-09-05| 723| 3| 
|  C  |2017-11-05|  23| 4| 
+------------+----------+-------+----+ 

私はこのランクを取得するための窓関数を使用しますが、私は私はこのような個別のテーブルを取得することができますどのように取得していないです。私はspark 2.0.0とjavaを使用しています。ここで

 WindowSpec ws = Window.partitionBy(Employee).orderBy(date); 
    Column rowNum = functions.row_number().over(ws);           
    data.withColumn("rank", rank().over(ws)) 

Answer 1

は、従業員の個別の値のためにフィルタリングすることによって、これを達成するためのサンプルコードです：

//Getting the distinct columns 
List<Row> distinctColumns = df.select("Employee").distinct().collectAsList(); 

//Initializing empty list for the new DataFrames 
ArrayList<Dataset<Row>> newDFs = new ArrayList<>(); 

WindowSpec ws = Window.orderBy("date"); 

//Filtering by the distinct column values and adding to the list. 
for (Row distinctColumn : distinctColumns) { 
    String colName = distinctColumn.getString(0); 

    newDFs.add(
      df.filter(col("Employee").$eq$eq$eq(colName)) 
        .withColumn("rank", rank().over(ws)) 
    ); 
} 

// show all the new DFs 
for (Dataset<Row> aDF : newDFs) { 
    aDF.show(); 
}