Spark: наиболее эффективный способ сортировки и разделения данных, которые должны быть записаны как паркет

Мои данные в принципе представляют собой таблицу, которая содержит столбец ID и столбец GROUP_ID, помимо других "данных".

На первом этапе я читаю CSV в Spark, делаю некоторую обработку для подготовки данных для второго шага и записываю данные как паркет. На втором этапе много groupBy('GROUP_ID') и Window.partitionBy('GROUP_ID').orderBy('ID').

Теперь цель - во избежание перетасовки на втором шаге - для эффективной загрузки данных на первом этапе, поскольку это один таймер.

Вопрос Часть 1: AFAIK, Spark сохраняет разбиение при загрузке с паркета (которое на самом деле является основой любого "оптимизированного рассмотрения записи" ) - правильно?

Я придумал три возможности:

• df.orderBy('ID').write.partitionBy('TRIP_ID').parquet('/path/to/parquet')
• df.orderBy('ID').repartition(n, 'TRIP_ID').write.parquet('/path/to/parquet')
• df.repartition(n, 'TRIP_ID').sortWithinPartitions('ID').write.parquet('/path/to/parquet')

Я бы установил n таким образом, чтобы отдельные файлы паркета составляли ~ 100 МБ.

Вопрос Часть 2: Правильно ли, что три варианта дают "одинаковые" /похожие результаты в отношении цели (избегайте перетасовки на втором шаге)? Если нет, в чем разница? И какой из них "лучше"?

Вопрос. Часть 3:. Какой из трех вариантов лучше работает в отношении шага 1?

Спасибо, что поделились своими знаниями!

EDIT 2017-07-24

После выполнения некоторых тестов (запись и чтение из паркета) кажется, что Spark не может восстановить partitionBy и orderBy информацию по умолчанию на втором шаге. Количество разделов (как получено из df.rdd.getNumPartitions(), по-видимому, определяется количеством ядер и/или spark.default.parallelism (если установлено), но не количеством паркетных разделов, поэтому ответ на вопрос 1 будет WRONG, а вопросы 2 и 3 будут неактуальны.

Итак, оказывается, что REAL QUESTION: есть ли способ сказать Spark, что данные уже разделены столбцом X и отсортированы по столбцу Y

Ответы

large.write
  .bucketBy(4, "id")
  .sortBy("id")
  .mode(SaveMode.Overwrite)
  .saveAsTable(bucketedTableName)