Получить значение для последней записи в случае нескольких записей для одной и той же группы - PullRequest
Новая
       118

Получить значение для последней записи в случае нескольких записей для одной и той же группы

1 голос
/ 19 июня 2020

У меня есть набор данных, в котором будет несколько записей для поля столбца id, сгруппированных по другим столбцам. Для этого набора данных я хочу создать новый столбец только для последней записи каждой группы. Я использовал case statement для получения нового столбца и union, чтобы получить значение для последней записи. Я думал избежать использования UNION, поскольку это дорогостоящая операция в spark-sql.

Ввод: 

person_id   order_id       order_ts         order_amt
   1            1       2020-01-01 10:10:10     10  
   1            2       2020-01-01 10:15:15     15  
   2            3       2020-01-01 10:10:10     0   
   2            4       2020-01-01 10:15:15     15

Из введенных выше данных person_id 1 имеет два заказа (1,2), а person_id 2 - два заказа (3,4). Я хочу получить столбец только для последнего заказа для данного человека.

Ожидаемый результат: 

person_id   order_id       order_ts         order_amt valid_order
   1            1       2020-01-01 10:10:10     10      N
   1            2       2020-01-01 10:15:15     15      Y
   2            3       2020-01-01 10:10:10     0       N
   2            4       2020-01-01 10:15:15     15      Y

Я попытался получить результат, используя UNION в запросе: 

select person_id, order_id, order_ts, order_amt, valid_order 
from 
(
select *, row_number() over(partition by order_id order by derive_order) as rnk 
from 
   (
   select person_id, order_id, order_ts, order_amt, 'N' as valid_order, 'before' as derive_order 
   from ips_core.test_table
   UNION
   select person_id, order_id, order_ts, order_amt, 
   case when order_amt is not null and order_amt >0 then 'Y' else 'N' end as valid_order, 
   'after' as derive_order 
   from 
      (
      select *, row_number() over(partition by person_id order by order_ts desc) as rnk 
      from ips_core.test_table
      ) where rnk = 1
   ) final 
) where rnk = 1 order by person_id, order_id;

Я также получил тот же результат, используя комбинация left outer join и inner join.

Запрос соединения: 

select final.person_id, final.order_id, final.order_ts, final.order_amt, 
case when final.valid_order is null then 'N' else final.valid_order end as valid_order 
from 
(
select c.person_id, c.order_id, c.order_ts, c.order_amt, d.valid_order from ips_core.test_table c 
left outer join 
  (
    select a.*, case when a.order_amt is not null and a.order_amt >0 then 'Y' else 'N' end as valid_order 
    from ips_core.test_table a 
    inner join 
    (
    select person_id, max(order_id) as order_id from ips_core.test_table group by 1
    ) b on a.person_id = b.person_id and a.order_id = b.order_id
  ) d on c.order_id = d.order_id
) final order by person_id, order_id;

Наш входной набор данных будет содержать около 20 миллионов записей. Есть ли более оптимизированный способ получить тот же результат, кроме приведенных выше запросов.

Любая помощь приветствуется.

Ответы [ 2 ]

1 голос
/ 19 июня 2020

Это может быть сделано без объединения или объединения. Также это условие a.order_amt is not null and a.order_amt >0 является избыточным, потому что, если сумма> 0, оно уже НЕ NULL. 

select person_id, order_id, order_ts, order_amt,
       case when rn=1 and order_amt>0 then 'Y' else 'N' end as valid_order
  from
      (
       select person_id, order_id, order_ts, order_amt,
              row_number() over(partition by person_id order by order_ts desc) as rn
         from ips_core.test_table a 
      ) s
1 голос
/ 19 июня 2020

проверьте, помогает ли - 

 val data =
      """
        |person_id  | order_id  |     order_ts         |order_amt
        |   1       |     1     |  2020-01-01 10:10:10 |    10
        |   1       |     2     |  2020-01-01 10:15:15 |    15
        |   2       |     3     |  2020-01-01 10:10:10 |    0
        |   2       |     4     |  2020-01-01 10:15:15 |    15
      """.stripMargin
    val stringDS = data.split(System.lineSeparator())
      .map(_.split("\\|").map(_.replaceAll("""^[ \t]+|[ \t]+$""", "")).mkString(","))
      .toSeq.toDS()
    val df = spark.read
      .option("sep", ",")
      .option("inferSchema", "true")
      .option("header", "true")
      .option("nullValue", "null")
      .csv(stringDS)
    df.printSchema()
    df.show(false)

    /**
      * root
      * |-- person_id: integer (nullable = true)
      * |-- order_id: integer (nullable = true)
      * |-- order_ts: timestamp (nullable = true)
      * |-- order_amt: integer (nullable = true)
      *
      * +---------+--------+-------------------+---------+
      * |person_id|order_id|order_ts           |order_amt|
      * +---------+--------+-------------------+---------+
      * |1        |1       |2020-01-01 10:10:10|10       |
      * |1        |2       |2020-01-01 10:15:15|15       |
      * |2        |3       |2020-01-01 10:10:10|0        |
      * |2        |4       |2020-01-01 10:15:15|15       |
      * +---------+--------+-------------------+---------+
      */

Использование Spark DSL 

    df.withColumn("latest", max($"order_ts").over(Window.partitionBy("person_id")))
      .withColumn("valid_order", when(unix_timestamp($"latest") - unix_timestamp($"order_ts") =!= 0, lit("N"))
          .otherwise(lit("Y"))
      )
      .show(false)

    /**
      * +---------+--------+-------------------+---------+-------------------+-----------+
      * |person_id|order_id|order_ts           |order_amt|latest             |valid_order|
      * +---------+--------+-------------------+---------+-------------------+-----------+
      * |2        |3       |2020-01-01 10:10:10|0        |2020-01-01 10:15:15|N          |
      * |2        |4       |2020-01-01 10:15:15|15       |2020-01-01 10:15:15|Y          |
      * |1        |1       |2020-01-01 10:10:10|10       |2020-01-01 10:15:15|N          |
      * |1        |2       |2020-01-01 10:15:15|15       |2020-01-01 10:15:15|Y          |
      * +---------+--------+-------------------+---------+-------------------+-----------+
      */

Использование SPARK SQL

// Spark SQL
    df.createOrReplaceTempView("order_table")
    spark.sql(
      """
        |select person_id, order_id, order_ts, order_amt, latest,
        | case when (unix_timestamp(latest) - unix_timestamp(order_ts) != 0) then 'N' else 'Y' end as  valid_order
        | from
        | (select person_id, order_id, order_ts, order_amt, max(order_ts) over (partition by person_id) as latest FROM order_table) a
      """.stripMargin)
      .show(false)

    /**
      * +---------+--------+-------------------+---------+-------------------+-----------+
      * |person_id|order_id|order_ts           |order_amt|latest             |valid_order|
      * +---------+--------+-------------------+---------+-------------------+-----------+
      * |2        |3       |2020-01-01 10:10:10|0        |2020-01-01 10:15:15|N          |
      * |2        |4       |2020-01-01 10:15:15|15       |2020-01-01 10:15:15|Y          |
      * |1        |1       |2020-01-01 10:10:10|10       |2020-01-01 10:15:15|N          |
      * |1        |2       |2020-01-01 10:15:15|15       |2020-01-01 10:15:15|Y          |
      * +---------+--------+-------------------+---------+-------------------+-----------+
      */
...