Первая и последняя функция PySpark над разделом в один go - PullRequest
Новая
       3

Первая и последняя функция PySpark над разделом в один go

2 голосов
/ 10 февраля 2020

У меня есть такой код pyspark, 

spark_df = spark_df.orderBy('id', 'a1', 'c1')
out_df = spark_df.groupBy('id', 'a1', 'a2').agg(
    F.first('c1').alias('c1'),
    F.last('c2').alias('c2'),
    F.first('c3').alias('c3'))

Мне нужно хранить данные в порядке порядка в идентификаторе заказа, a1 и c1. Затем выберите столбцы, как показано выше, для группы, определенной для ключей id, a1 и c1.

Из-за первого и последнего недетерминизма я изменил код на этот уродливый код, который работает, но я не уверен это эффективно. 

w_first = Window.partitionBy('id', 'a1', 'a2').orderBy('c1')
w_last = Window.partitionBy('id', 'a1', 'a2').orderBy(F.desc('c1'))

out_first = spark_df.withColumn('Rank_First', F.rank().over(w_first)).filter(F.col('Rank_First') == 1).drop(
    'Rank_First')
out_last = spark_df.withColumn('Rank_Last', F.rank().over(w_last)).filter(F.col('Rank_First') == 1).drop(
    'Rank_Last')

out_first = out_first.withColumnRenamed('c1', 'First_c1') \
    .withColumnRenamed('c2', 'First_c2') \
    .withColumnRenamed('c3', 'First_c3')

out_last = out_last.withColumnRenamed('c1', 'Last_c1') \
    .withColumnRenamed('c2', 'Last_c2') \
    .withColumnRenamed('c3', 'Last_c3')

out_df = out_first.join(out_last, ['id', 'a1', 'a2']) \
    .select('id', 'a1', 'a2', F.col('First_c1').alias('c1'),
            F.col('Last_c2').alias('c2'),
            F.col('First_c3').alias('c3'))

Я пытался найти лучшую и эффективную альтернативу. Я бегу к bottle шеям производительности, когда размер данных огромен.

Есть ли лучшая альтернатива сделать первое и последнее через окно, упорядоченное в определенном порядке c в одном go.

1 Ответ

2 голосов
/ 10 февраля 2020

Вы все еще можете использовать функции last и first над окном, которое гарантирует детерминизм. Вам нужно добавить границы с rowsBetween при указании окна, чтобы last давал правильные значения (согласно этому post ).

Попробуйте: 

w = Window.partitionBy('id', 'a1', 'a2').orderBy('c1') \
          .rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)

df = df.withColumn("First_c1", first("c1").over(w)) \
      .withColumn("First_c3", first("c3").over(w)) \
      .withColumn("Last_c2", last("c2").over(w))

df.groupby("id", "a1", "a2")\
  .agg(first("First_c1").alias("c1"),
       first("Last_c2").alias("c2"),
       first("First_c3").alias("c3")
  ).show()
...