Каков наилучший способ реализации ROW_NUMBER (генератор последовательности) в программе Spark для миллиардов записей (> 25 миллиардов)?
Пример кода:
select patient_id,
department_id,
row_number() over (partition by department_id order by dept_id asc) as Pat_serial_Nbr
from T_patient;
Row_number () в программе Spark работает более 4 часов и завершается с ошибкой для 15 миллиардов записей.
Я пробовал метод RDD zipWithIndex () для 15 миллиардов записей (выполнение заняло 40 минут), он возвращает ожидаемые результаты.
public RDD<scala.Tuple2<T,Object>> zipWithIndex()
Сжатие этого СДР с индексами его элементов. Порядок сначала определяется на основе индекса раздела, а затем порядка элементов в каждом разделе. Таким образом, первый элемент в первом разделе получает индекс 0, а последний элемент в последнем разделе получает самый большой индекс. Это похоже на zipWithIndex в Scala, но в качестве типа индекса используется Long вместо Int. Этот метод должен запускать искровое задание, когда этот СДР содержит более одного раздела.
Обратите внимание, что некоторые RDD, такие как возвращаемые groupBy (), не гарантируют порядок элементов в разделе. Следовательно, индекс, присвоенный каждому элементу, не гарантируется и может даже измениться, если СДР будет переоценен. Если для обеспечения одинаковых назначений индекса требуется фиксированный порядок, следует отсортировать СДР с помощью sortByKey () или сохранить его в файл.
scala> List("X", "Y", "Z").zipWithIndex
res0: List[(String, Int)] = List((X,0), (Y,1), (Z,2))
То же самое работает для фрейма данных - val rdd = df.rdd.zipWithIndex
Второй вариант:
val df_id = df.withColumn("id",monotonicallyIncreasingId)
Ссылка: Spark-монотонно увеличивающийся идентификатор не работает должным образом в кадре данных?
Не могли бы вы предложить оптимальный способ генерации порядковых номеров в Scala Spark.