Spark SQL 中的分组和聚合元组

Question

我正在使用 Spark SQL 和 Java。我有一个数据集，其中包含按 ENTITY 和 DOCUMENT_ID 分组的重复客户端。我添加了一个 rownumber 列来了解每个组有多少客户（我必须比较）：

.withColumn( "ROWNUMBER", row_number().over(Window.partitionBy("ENTITY", "ENTITY_DOC").orderBy("ID")))

+---------+----------+-----------------+----------+-----------+-----------+--------+------------+
|ROWNUMBER|  ENTITY  |      ENTITY_DOC |    ID    |  BLOCK    |  TYPE_DOC |COD_BEST|COD_CRITERIO|
+---------+----------+-----------------+----------+-----------+-----------+--------+------------+
|        1|       182|000004693R       |   5254578|          3|         01|       0|           0|
|        2|       182|000004693R       |  99841470|          0|         01|       0|           0|
|        3|       182|000004693R       |  45866239|          3|         01|       0|           0|
|        1|       182|000081638B       |  99804050|          0|         01|       0|           0|
|        2|       182|000081638B       |  99803968|          0|         01|       0|           0|
|        3|       182|000081638B       |  99803958|          0|         01|       0|           0|
|        4|       182|000081638B       |  99804054|          0|         01|       0|           0|
|        5|       182|000081638B       |  99787706|          1|         01|       0|           0|
|        6|       182|000081638B       |  99803930|          0|         01|       0|           0|
|        1|       182|000107084L       |  99819126|          0|         01|       0|           0|
|        2|       182|000107084L       |  99818446|          0|         01|       0|           0|
+---------+----------+-----------------+----------+-----------+-----------+--------+------------+

现在我必须比较成对的行以确定哪个是最好的。

First compare rownumber1 vs rownumber2 (if rownumber2 is the best) then
compare rownumber2 vs rownumber3 (if rownumber3 is the best)
then compare rownumber3 vs rownumber4 ... etc
 

根据某些业务标准决定哪个是最好的，例如：

//criteria 1
BLOCK = 1 VS BLOCK 1
   //go to the next criteria
//criteria 2
BLOCK = 2 VS BLOCK 1
   //the best is BLOCK 2

//criteria3
TYPE_DOC = 1 VS TYPE_DOC = 1
 //go to the next criteria

//criteria4
 TYPE_DOC = 1 VS TYPE_DOC = 2
  //the best is TYPE_DOC 1

(not is a logical example but to get an idea)

最后我必须知道哪一行是每个组中最好的行以及它被选中的标准，但我不知道如何迭代每个组以比较其行的字段。

会不会很困难？

Answer 1

您可以先为每个重复项分配row_number，然后从列创建一个映射并将行号添加为每个键的后缀。最后按ENTITY、DOCUMENT_ID 分组并收集地图列表，将它们连接起来并在地图爆炸后进行旋转。

请注意，这里我主要使用 SQL 表达式，因为我对 Spark Java API 不是很熟悉，但如果你想将它们转换为使用 API 函数，逻辑是相同的。

Dataset < Row > tuples = duplicates.withColumn(
    "rn",
    row_number().over(Window.partitionBy("ENTITY", "DOCUMENT_ID").orderBy("ID"))
).withColumn(
    "dupes",
    expr("map(concat('COUNTRY_', rn), COUNTRY, concat('ID_', rn), ID, concat('CUSTOMER_NAME_', rn), CUSTOMER_NAME)")
).groupBy("ENTITY", "DOCUMENT_ID").agg(
    collect_list("dupes").alias("dupes")
).selectExpr(
    "ENTITY",
    "DOCUMENT_ID",
    "explode(aggregate(dupes, cast(map() as map<string,string>), (acc, x) -> map_concat(acc, x)))"
).groupBy(
    "ENTITY", "DOCUMENT_ID"
).pivot("key").agg(first("value"));


tuples.show();

//+------+-----------+---------+---------+---------------+---------------+--------+--------+
//|ENTITY|DOCUMENT_ID|COUNTRY_1|COUNTRY_2|CUSTOMER_NAME_1|CUSTOMER_NAME_2|    ID_1|    ID_2|
//+------+-----------+---------+---------+---------------+---------------+--------+--------+
//|    11|  A06804173|        9|        9|     Elton John|     Elton John|12341000|13701921|
//+------+-----------+---------+---------+---------------+---------------+--------+--------+

---

另一种方法是按您的键列分组并收集结构列表，然后使用结果数组的最大大小，访问元素并创建多个列。像这样的：

import java.util.stream.IntStream;
import java.util.stream.Stream;

Dataset<Row> tuples = duplicates.groupBy("ENTITY", "DOCUMENT_ID").agg(
    collect_list(expr("struct(COUNTRY, ID, CUSTOMER_NAME)")).alias("dupes")
);

int maxSize = tuples.select(max(size(col("dupes")))).first().getInt(0);

Column[] dupes = IntStream.rangeClosed(0, maxSize - 1)
        .mapToObj(i -> new Column[]{
                col("dupes").getItem(i).getField("COUNTRY").alias("COUNTRY_" + i),
                col("dupes").getItem(i).getField("ID").alias("ID_" + i),
                col("dupes").getItem(i).getField("CUSTOMER_NAME").alias("CUSTOMER_NAME_" + i),
        }).flatMap(Stream::of).toArray(Column[]::new);

tuples.select(
    Stream.of(new Column[]{col("ENTITY"), col("DOCUMENT_ID")}, dupes)
            .flatMap(Stream::of).toArray(Column[]::new)
).show();