Вторичная сортировка в искре

Вторичная сортировка — это метод, который позволяет упорядочивать по значению (значениям) (в дополнение к сортировке по ключу) на этапе сокращения задания Map-Reduce. Например, вы можете захотеть anyalize вход пользователя в ваше приложение. Сортировка результатов по дню и времени, а также идентификатор пользователя (естественный ключ) помогут определить тенденции пользователей. Дополнительный порядок по дню и времени является примером вторичной сортировки. В то время как я писал ранее о  вторичной сортировке в Hadoop , эта статья расскажет, как мы выполняем вторичную сортировку в Spark.

Настроить

Мы будем использовать RDD API для реализации вторичной сортировки. Хотя это может быть легко достигнуто с помощью  DataFrames , мы не будем останавливаться на этом подходе. Данные, которые мы используем, — это данные о своевременной работе авиакомпании в  Бюро транспорта . Доступно несколько точек данных, но мы сосредоточимся на том, какие авиакомпании имеют наиболее поздние прибытия и в каком аэропорту происходят такие поздние прибытия. Из этого утверждения мы можем определить наш порядок сортировки: идентификатор авиакомпании, идентификатор аэропорта и время задержки. Отказ от ответственности:  Этот пример только для демонстрационных целей! Это не предназначено, чтобы вывести или определить фактическую работу авиакомпании.

Создание пар ключ-значение

Данные загружаются в формате CSV и будут преобразованы в формат ключ-значение. Важной частью вторичной сортировки является то, какие значения включить в ключ, чтобы включить дополнительный порядок. «Естественный» ключ — это AirlinesId, а ArriAirportId и birthyDelay — значения, которые мы включим в ключ. Это представлено  FlightKey классом дела:

case class FlightKey(airLineId: String, arrivalAirPortId: Int, arrivalDelay: Double)

Наша первая попытка создания наших пар ключ-значение выглядит следующим образом:

val rawDataArray = sc.textFile(args(0)).map(line => line.split(","))
  //Using keyBy but retains entire array in value
  val keyedByData = rawDataArray.keyBy(arr => createKey(arr))

 //supporting code
  def createKey(data: Array[String]): FlightKey = {
    FlightKey(data(UNIQUE_CARRIER), safeInt(data(DEST_AIRPORT_ID)), safeDouble(data(ARR_DELAY)))
  }

val rawDataArray = sc.textFile(args(0)).map(line => line.split(","))
 val airlineData = rawDataArray.map(arr => createKeyValueTuple(arr))

 //supporting code
 def createKeyValueTuple(data: Array[String]) :(FlightKey,List[String]) = {
      (createKey(data),listData(data))
  }

  def createKey(data: Array[String]): FlightKey = {
    FlightKey(data(UNIQUE_CARRIER), safeInt(data(DEST_AIRPORT_ID)), safeDouble(data(ARR_DELAY)))
  }

  def listData(data: Array[String]): List[String] = {
    List(data(FL_DATE), data(ORIGIN_AIRPORT_ID), data(ORIGIN_CITY_MARKET_ID), data(DEST_CITY_MARKET_ID))
  }

class AirlineFlightPartitioner(partitions: Int) extends Partitioner {
    require(partitions >= 0, s"Number of partitions ($partitions) cannot be negative.")

    override def numPartitions: Int = partitions

    override def getPartition(key: Any): Int = {
      val k = key.asInstanceOf[FlightKey]
      k.airLineId.hashCode() % numPartitions
    }
  }

object FlightKey {
    implicit def orderingByIdAirportIdDelay[A <: FlightKey] : Ordering[A] = {
       Ordering.by(fk => (fk.airLineId, fk.arrivalAirPortId, fk.arrivalDelay * -1))
    }
  }

object SecondarySort extends SparkJob {

  def runSecondarySortExample(args: Array[String]): Unit = {

    val sc = context("SecondarySorting")
    val rawDataArray = sc.textFile(args(0)).map(line => line.split(","))
    val airlineData = rawDataArray.map(arr => createKeyValueTuple(arr))

    val keyedDataSorted = airlineData.repartitionAndSortWithinPartitions(new AirlineFlightPartitioner(1))

    //only done locally for demo purposes, usually write out to HDFS
    keyedDataSorted.collect().foreach(println)
  }
}

(FlightKey("AA",10397,-2.0),List(2015-01-01, 11298, 30194, 30397))
(FlightKey("AA",11278,-2.0),List(2015-01-01, 11298, 30194, 30852))
(FlightKey("AA",11278,-14.0),List(2015-01-01, 12892, 32575, 30852))
(FlightKey("AA",11292,24.0),List(2015-01-01, 13930, 30977, 30325))
(FlightKey("AA",11298,133.0),List(2015-01-01, 13891, 32575, 30194))
(FlightKey("AA",11298,109.0),List(2015-01-01, 12173, 32134, 30194))
(FlightKey("AA",11298,55.0),List(2015-01-01, 14107, 30466, 30194))
(FlightKey("AA",11298,49.0),List(2015-01-01, 12478, 31703, 30194))
(FlightKey("AA",11298,40.0),List(2015-01-01, 14771, 32457, 30194))
(FlightKey("AA",11298,35.0),List(2015-01-01, 12094, 34699, 30194))
(FlightKey("AA",11298,24.0),List(2015-01-01, 13830, 33830, 30194))
(FlightKey("AA",11298,23.0),List(2015-01-01, 14869, 34614, 30194))
(FlightKey("AA",11298,19.0),List(2015-01-01, 11433, 31295, 30194))
(FlightKey("AA",11298,17.0),List(2015-01-01, 12264, 30852, 30194))