Spark Dataframes — сокращение по ключу

Допустим, у меня есть подобная структура данных, где ts - это временная метка.

case class Record(ts: Long, id: Int, value: Int)

Учитывая большое количество этих записей, я хочу получить запись с самой высокой отметкой времени для каждого идентификатора. Используя API RDD, я думаю, что следующий код выполняет свою работу:

def findLatest(records: RDD[Record])(implicit spark: SparkSession) = {
  records.keyBy(_.id).reduceByKey{
    (x, y) => if(x.ts > y.ts) x else y
  }.values
}

Точно так же это моя попытка с наборами данных:

def findLatest(records: Dataset[Record])(implicit spark: SparkSession) = {
  records.groupByKey(_.id).mapGroups{
    case(id, records) => {
      records.reduceLeft((x,y) => if (x.ts > y.ts) x else y)
    }
  }
}

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

records.groupBy($"id")

Но это дает мне RelationGroupedDataSet, и мне не ясно, какую функцию агрегации мне нужно написать, чтобы достичь того, чего я хочу — все примеры агрегаций, которые я видел, сосредоточены на возврате только одного агрегируемого столбца, а не всей строки.

Можно ли добиться этого с помощью фреймов данных?


apache-spark apache-spark-sql scala apache-spark-dataset
person d80tb7    schedule 20.12.2016    source источник

Ответы (2)


arrow_upward
12
arrow_downward

Вы можете использовать логику argmax (см. пример)

Например, предположим, что ваш фрейм данных называется df и имеет идентификаторы столбцов, val, ts, вы бы сделали что-то вроде этого:

import org.apache.spark.sql.functions._
val newDF = df.groupBy('id).agg.max(struct('ts, 'val)) as 'tmp).select($"id", $"tmp.*")
person Assaf Mendelson    schedule 20.12.2016
comment
да, это работает отлично, спасибо! кроме того, я думаю, что у вас есть небольшая опечатка в ответе - точка после agg должна быть скобкой! - person d80tb7; 20.12.2016
comment
Можете ли вы объяснить, почему это работает? Максимум применяется только к первому столбцу? - person user238607; 22.08.2017
comment
@ user238607 max использует порядок столбцов. Неявный порядок для структуры (или кортежа) определяется порядком первого элемента. - person Assaf Mendelson; 22.08.2017
comment
работает нормально, согласен с ТС, есть опечатка - должно быть ( между agg & max - person hooke; 06.04.2020

arrow_upward
0
arrow_downward

Для наборов данных я сделал это, протестировал на Spark 2.1.1.

final case class AggregateResultModel(id: String,
                                      mtype: String,
                                      healthScore: Int,
                                      mortality: Float,
                                      reimbursement: Float)
.....
.....

// assume that the rawScores are loaded behorehand from json,csv files

val groupedResultSet = rawScores.as[AggregateResultModel].groupByKey( item => (item.id,item.mtype ))
      .reduceGroups( (x,y) => getMinHealthScore(x,y)).map(_._2)


// the binary function used in the reduceGroups

def getMinHealthScore(x : AggregateResultModel, y : AggregateResultModel): AggregateResultModel = {
    // complex logic for deciding between which row to keep
    if (x.healthScore > y.healthScore) { return y }
    else if (x.healthScore < y.healthScore) { return x }
    else {

      if (x.mortality < y.mortality) { return y }
      else if (x.mortality > y.mortality) { return x }
      else  {

        if(x.reimbursement < y.reimbursement)
          return x
        else
          return y

      }

    }

  }
person user238607    schedule 24.08.2017