Блог

У меня есть требование объединить 5 таблиц среднего размера (~ 80 гб каждая) с большим объемом входных данных ~ 800 Гб. Все данные хранятся в таблицах HIVE. Для достижения этой цели я использую Spark SQL 1.6.1. Для завершения объединения требуется 40 минут времени --num-executors 20 --driver-memory 40g --executor-memory 65g --executor-cores 6 . Все соединения являются внешними объединениями типа sort merge. Также происходит много перетасовки.

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

Есть ли какой-либо другой способ, которым я могу физически разделить и отсортировать данные в Hive (без файлов частей), чтобы spark знал о ключах разделения при загрузке данных из самого hive и выполнял объединение с тем же разделением без перетасовки данных? Это позволит избежать дополнительного повторного разделения после загрузки данных из hive.

Прежде всего, Spark Sql 1.6.1 пока не поддерживает сегменты hive. Таким образом, в этом случае мы остаемся с операциями уровня Spark, гарантирующими, что все таблицы должны переходить в одни и те же разделы spark при загрузке данных. Spark API предоставляет перераспределение и сортировку по частям для достижения того же. например

val part1 = df1.repartition(df1(«key1»)).sortWithinPartitions(df1(«key1»))

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

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