基于另一列范围的子集中值的排名

Question

我有以下 DataFrame（称之为dfTest）：

item	position	value
a	1	8
b	2	9
c	3	10
d	4	7
e	5	9

我要做的是，给定距离限制1，找出距离（由@987654326 定义的行子集内的每一行中value 的排名索引（按降序排序） @column) 到这个item 不大于1。

例如，所需的输出（称为dfResult）应该是：

item	position	value	rank
a	1	8	2
b	2	9	2
c	3	10	1
d	4	7	3
e	5	9	1

解释：

1. 对于a 行。由于b 是唯一一个与a 的距离abs(2-1)=1) 的item，所以value 的排序子集是[9,8] 和a 的value 的顺序是第二位，因此a 的rank 为2。
2. 对于b 行。这里，a 和 c 是距离到 b items，value 的排序子集是 [10,9,8] 和 b 的 value 是第二个，因此b 的rank 是2。
3. 对于c 行。这里，b 和 d 是距离到 c items，value 的排序子集是 [10,9,7] 和 c 的 value 是第一个，因此c 的rank 为1。
4. ...

感谢 Spark SQL、Scala 或 PySpark 中的任何解决方案！

Answer 1

您可以使用自联接来确定整个表中的相关项（例如，如果您有一个位置为2 的项f，a 的排名可能会改变）和rank窗口函数来达到你想要的排名。

使用 spark-sql、pyspark 和 scala 的示例如下：

使用 spark-sql

SELECT
    d.item,
    d.position,
    d.value,
    d.rank
FROM (
    SELECT
        i1.*,
        i2.item as other_item,
        RANK() OVER (
            PARTITION BY i1.item
            ORDER BY i2.value DESC
        ) as rank
    FROM
        dfTest i1
    INNER JOIN
        dfTest i2 ON ABS(i1.position-i2.position)<=1
) d
WHERE item=other_item;

View working demo on db-fiddle

使用 pyspark API

from pyspark.sql import functions as F
from pyspark.sql import Window

dfResult = (
    dfTest.alias("i1")
          .join(
              dfTest.alias("i2"),
              F.abs(
                  F.col("i1.position")-F.col("i2.position")
              ) <=1,
              "inner"
          )
          .withColumn(
              "rank",
              F.rank().over(
                  Window.partitionBy("i1.item").orderBy(F.col("i2.value").desc())
              )
          )
          .where(F.col("i1.item")==F.col("i2.item"))
          .select("i1.*","rank")
          
)

使用斯卡拉

val dfResult = 
    dfTest.alias("i1")
          .join(
              dfTest.alias("i2"),
              abs(
                  col("i1.position")-col("i2.position")
              ) <=1,
              "inner"
          )
          .withColumn(
              "rank",
              rank().over(
                  Window.partitionBy("i1.item").orderBy(col("i2.value").desc())
              )
          )
          .where(col("i1.item")==col("i2.item"))
          .select("i1.*","rank")
          

输出

dfResult.show(truncate=False)

+----+--------+-----+----+
|item|position|value|rank|
+----+--------+-----+----+
|   a|       1|    8|   2|
|   b|       2|    9|   2|
|   c|       3|   10|   1|
|   d|       4|    7|   3|
|   e|       5|    9|   1|
+----+--------+-----+----+

让我知道这是否适合你。