Перекошенный в Искре

Question

У меня есть набор данных, который я хочу разделить по определенному ключу (clientID), но некоторые клиенты выдают гораздо больше данных, чем другие. В Hive есть функция под названием « ListBucketing », вызываемая « skewed by » специально для решения этой ситуации.

Однако я не могу найти никаких признаков того, что Spark поддерживает эту функцию или как (если она поддерживает) ее использовать.

Есть ли функция Spark, которая является эквивалентной? Или у Spark есть какой-то другой набор функций, с помощью которых можно воспроизвести это поведение?

(В качестве бонуса - и требования для моего реального варианта использования - работает ли предложенный вами метод с Amazon Athena?)

Answer 1

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

Допустим, вы хотите разделить по столбцу "y", но данные очень искажены, как в этом игрушечном примере(1 раздел с 5 строками, остальные с одной строкой):

val df = spark.range(8).withColumn("y", when('id < 5, 0).otherwise('id))
df.show()
+---+---+
| id|  y|
+---+---+
|  0|  0|
|  1|  0|
|  2|  0|
|  3|  0|
|  4|  0|
|  5|  5|
|  6|  6|
|  7|  7|
+-------+

Теперь давайте добавим искусственный случайный столбец и напишем фрейм данных.

val maxNbOfBuckets = 3
val part_df = df.withColumn("r", floor(rand() * nbOfBuckets))
part_df.show
+---+---+---+
| id|  y|  r|
+---+---+---+
|  0|  0|  2|
|  1|  0|  2|
|  2|  0|  0|
|  3|  0|  0|
|  4|  0|  1|
|  5|  5|  2|
|  6|  6|  2|
|  7|  7|  1|
+---+---+---+

// and writing. We divided the partition with 5 elements into 3 partitions.
part_df.write.partitionBy("y", "r").csv("...")