Как Spark может записывать большие файлы без дополнительных данных?

Question

Я использую Spark EMR для обработки данных и записи их в S3. Данные разбиты по дате. В случае, когда мы повторно обрабатываем те же данные даты, я использую специальную функцию, которая сравнивает текущий обработанный кадр данных с данными, которые уже находятся в S3. Обе данные объединены, поэтому данные не теряются.

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

Первая запись приводит к 200 файлам переменного размера (20-100 КБ) на общую сумму 74 МБ. Вторая запись приводит к 200 файлам фиксированного размера (около 430 КБ каждый) на общую сумму 84 МБ.

Я сравнил обе данные из разных записей, импортировав их в кадры данных, количество строк аналогично. Данные одинаковы (я использовал df1.exceptAll (df2)).

Почему разница в размерах файлов между первой и второй записью? Откуда взялись эти дополнительные 10 МБ?

Я не использую перераспределения / объединения.

Заранее спасибо.

Answer 1

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

Образец:

import spark.implicits._
val df1 = Seq(
  (1,2,3),
  (4,5,6)
).toDF("col_a", "col_b", "col_c")

val df2 = Seq(
  (1,2,3),
  (4,5,6),
  (4,5,6)
).toDF("col_a", "col_b", "col_c")

df1.show()
df2.show()


// output:

+-----+-----+-----+
|col_a|col_b|col_c|
+-----+-----+-----+
|    1|    2|    3|
|    4|    5|    6|
+-----+-----+-----+

+-----+-----+-----+
|col_a|col_b|col_c|
+-----+-----+-----+
|    1|    2|    3|
|    4|    5|    6|
|    4|    5|    6|
+-----+-----+-----+

exceptAll проверки:

df1.exceptAll(df2).show()

+-----+-----+-----+
|col_a|col_b|col_c|
+-----+-----+-----+
+-----+-----+-----+

df2.exceptAll(df1).show()

+-----+-----+-----+
|col_a|col_b|col_c|
+-----+-----+-----+
|    4|    5|    6|
+-----+-----+-----+