Scalaを使用してSparkでスライディングウィンドウランクを設定するにはどうすればよいですか？

私はデータセットを持っています：

+-----+-------------------+---------------------+------------------+
|query|similar_queries    |model_score          |count             |
+-----+-------------------+---------------------+------------------+
|shirt|funny shirt        |0.0034038130658784866|189.0             |
|shirt|shirt womens       |0.0019435265241921438|136.0             |
|shirt|watch              |0.001097496453284101 |212.0             |
|shirt|necklace           |6.694577024597908E-4 |151.0             |
|shirt|white shirt        |0.0037413097560623485|217.0             |
|shirt|shoes              |0.0022062579255572733|575.0             |
|shirt|crop top           |9.065831060804897E-4 |173.0             |
|shirt|polo shirts for men|0.007706416273211698 |349.0             |
|shirt|shorts             |0.002669621942466027 |200.0             |
|shirt|black shirt        |0.03264296242546658  |114.0             |
+-----+-------------------+---------------------+------------------+

最初に「カウント」に基づいてデータセットをランク付けしています。

lazy val countWindowByFreq = Window.partitionBy(col(QUERY)).orderBy(col(COUNT).desc)
val ranked_data = data.withColumn("count_rank", row_number over countWindowByFreq)

+-----+-------------------+---------------------+------------------+----------+
|query|similar_queries    |model_score          |count             |count_rank|
+-----+-------------------+---------------------+------------------+----------+
|shirt|shoes              |0.0022062579255572733|575.0             |1         |
|shirt|polo shirts for men|0.007706416273211698 |349.0             |2         |
|shirt|white shirt        |0.0037413097560623485|217.0             |3         |
|shirt|watch              |0.001097496453284101 |212.0             |4         |
|shirt|shorts             |0.002669621942466027 |200.0             |5         |
|shirt|funny shirt        |0.0034038130658784866|189.0             |6         |
|shirt|crop top           |9.065831060804897E-4 |173.0             |7         |
|shirt|necklace           |6.694577024597908E-4 |151.0             |8         |
|shirt|shirt womens       |0.0019435265241921438|136.0             |9         |
|shirt|black shirt        |0.03264296242546658  |114.0             |10        |
+-----+-------------------+---------------------+------------------+----------+

現在、row_number（4行）のローリングウィンドウを使用してコンテンツをランク付けし、model_scoreに基づいてウィンドウ内でランク付けしようとしています。例：

最初のウィンドウ、row_number 1から4では、新しいランク（新しい列）は次のようになります。

1. polo shirts for men
2. white shirt
3. shoes
4. watch

最初のウィンドウ、row_number 5から8では、新しいランク（新しい列）は次のようになります。

5. funny shirt
6. shorts
7. shirt womens 
8. crop top

最初のウィンドウで、row_number 9を休ませると、新しいランク（新しい列）は次のようになります。

9. black shirt 
10. shirt womens

このスパークとScalaでどのように達成できるか教えてもらえますか？使用できる定義済みの関数はありますか？

私は試した ：

怠惰なvalMODEL_RANK = Window.partitionBy（col（QUERY））。orderBy（col（MODEL_SCORE）.desc）.rowsBetween（0、3）

しかし、これは私に与えます：

sql.AnalysisException: Window Frame ROWS BETWEEN CURRENT ROW AND 3 FOLLOWING must match the required frame ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW;

また、.rowsBetween（-3、0）で試してみましたが、エラーも発生します：

org.apache.spark.sql.AnalysisException: Window Frame ROWS BETWEEN 3 PRECEDING AND CURRENT ROW must match the required frame ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW;