Что такое копия на записи?

Отказ от ответственности :

Этот ответ содержит в основном предположения. Детальное объяснение этого явления может потребовать углубленного анализа входных и выходных данных (или, по крайней мере, их соответствующих метаданных).

Наблюдения :

1. Энтропия эффективно ограничивает производительность максимально возможного сжатия без потерь - Википедия - Энтропия (теория информации) [1117 ]. [1 123]

2. Оба постоянных столбчатых формата , а также внутреннее представление Spark SQL прозрачно применяют различные методы сжатия (например, кодирование по длине прогона или . ] словарь кодирования ), чтобы уменьшить объем памяти хранимых данных.

   Кроме того, форматы дисков (включая текстовые данные) могут быть явно сжаты с использованием алгоритмов сжатия общего назначения - неясно, так ли это здесь.

3. Сжатие (явное или прозрачное) применяется к блокам данных (обычно это разделы, но могут использоваться меньшие единицы).

4. На основании 1), 2) и 3) можно предположить, что средняя степень сжатия будет зависеть от распределения данных в кластере. Мы также должны отметить, что конечный результат может быть недетерминированным, если линия вверх по течению содержит широкие преобразования.

Возможное влияние coalesce против repartition :

В целом coalesce может идти двумя путями:

[ 1161]

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

val df = sc.parallelize(
  Seq("foo", "foo", "foo", "bar", "bar", "bar"),
  6 
).toDF

Если бы такой набор данных был записан на диск, не было бы возможности для сжатия - каждое значение должно быть записано как есть:

df.withColumn("pid", spark_partition_id).show

+-----+---+
|value|pid|
+-----+---+
|  foo|  0|
|  foo|  1|
|  foo|  2|
|  bar|  3|
|  bar|  4|
|  bar|  5|
+-----+---+

Другими словами, нам нужно примерно 6 * 3 байта, что дает всего 18 байтов.

Однако, если мы объединим

df.coalesce(2).withColumn("pid", spark_partition_id).show

+-----+---+
|value|pid|
+-----+---+
|  foo|  0|
|  foo|  0|
|  foo|  0|
|  bar|  1|
|  bar|  1|
|  bar|  1|
+-----+---+

, мы можем, например, применить RLE с маленьким int в качестве счетчика и сохранить каждый раздел 3 + 1 байт, дающий в общей сложности 8 байт.

Это, конечно, огромное упрощение, но показывает, как сохранение структуры входов с низкой энтропией и объединение блоков может привести к уменьшению использования памяти.

Второй coalesce сценарий менее очевиден, но есть сценарии, в которых энтропия может быть уменьшена восходящим процессом (подумайте, например, о оконных функциях), и сохранение такой структуры будет полезным.

Как насчет repartition ?

Без выражения разделения repartition применяется RoundRobinPartitioning (реализовано как HashPartitioning с псевдослучайным ключом на основе раздела Я бы). Пока хеш-функция ведет себя разумно, такое перераспределение должно максимизировать энтропию данных и в результате уменьшить возможную степень сжатия.

Заключение :

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

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

Наконец, repartition с выражением разделения или repartitionByRange должны уменьшить энтропию и улучшить степень сжатия.

Примечание :

Следует также помнить, что столбчатые форматы обычно выбирают конкретный метод сжатия / кодирования (или его отсутствие) на основе статистики времени выполнения. Таким образом, даже если набор строк в определенном блоке является фиксированным, но порядок строк изменяется, мы можем наблюдать различные результаты.