Как эффективно соединить очень большой стол и большой стол в Pyspark

Не уверен насчет памяти вашего драйвера и исполнителя, но в целом возможны две оптимизации соединения - широковещательная передача небольшой таблицы всем исполнителям и наличие одного и того же ключа раздела для обоих фреймов данных. В вашем случае перераспределение на основе вашего идентификатора поиска сделает его быстрее, если таблица 2 слишком велика для трансляции. Но ремонт имеет свою цену. Вы можете найти больше здесь - https://umbertogriffo.gitbook.io/apache-spark-best-practices-and-tuning/avoiding_shuffle_less_stage-_more_fast#:%7E:text=One%20way%20to%20avoid.%20shuffles,затем%20broadcast%20to%20every%20executor.

Дайте мне знать, что вы думаете. с нетерпением жду обсуждения в этой теме.

Если вы не можете транслировать, пример избегания присоединения с использованием группирования - вдохновленный здесь: Spark: предотвращение перемешивания / обмена при объединении двух идентично разделенных фреймов данных

spark.catalog.setCurrentDatabase(<your databasename>)
test1.write.mode('overwrite').bucketBy(100,'item').saveAsTable('table_item')
test2.write.mode('overwrite').bucketBy(100,'item').saveAsTable('table_item1')
#test1.

#%%
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1) # this is just to disable auto broadcasting for testing
import pyspark.sql.functions as F
inputDf1 = spark.sql("select * from table_item")
inputDf2 = spark.sql("select * from table_item1")
inputDf3 = inputDf1.alias("df1").join(inputDf2.alias("df2"),on='item')

Теперь попробуй

inputDf3.explain()

Результат будет примерно таким:

== Physical Plan ==
*(3) Project [item#1033, col1#1030, col2#1031, col3#1032, id#1038]
+- *(3) SortMergeJoin [item#1033], [item#1039], Inner
   :- *(1) Sort [item#1033 ASC NULLS FIRST], false, 0
   :  +- *(1) Project [col1#1030, col2#1031, col3#1032, item#1033]
   :     +- *(1) Filter isnotnull(item#1033)
   :        +- *(1) FileScan parquet 
   +- *(2) Sort [item#1039 ASC NULLS FIRST], false, 0
      +- *(2) Project [id#1038, item#1039]
         +- *(2) Filter isnotnull(item#1039)
            +- *(2) FileScan parquet 

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

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

Конкретный ответ на это ... Если у вас есть несжатый текстовый файл размером 30 ГБ, хранящийся в HDFS, то при настройке размера блока HDFS по умолчанию (128 МБ) он будет храниться в 235 блоках, что означает, что RDD, который вы читаете из этого файла, будет 235 разделов.

Здесь есть две вещи 1. Плоские файлы, например, CSV, и 2. сжатый файл, например паркет.

1. Когда у вас есть текстовый файл ... Когда Spark читает файл из HDFS, он создает один раздел для одного входного разделения. Разделение ввода устанавливается Hadoop InputFormat, используемым для чтения этого файла. Например, если вы используете textFile (), это будет TextInputFormat в Hadoop, который вернет вам один раздел для одного блока HDFS (но разделение между разделами будет выполняться по разделению строк, а не по точному разделению блока), если только у вас есть сжатый текстовый файл.

2. Для паркетного или сжатого файла: в случае сжатого файла вы получите один раздел для одного файла (поскольку сжатые текстовые файлы не могут быть разделены).

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

Итак, ответьте: Вопрос 1. Как распараллелить сканирование table_1 на основе разделов create_date? Это уже разделено

Для, Вопрос 2: есть ли другой способ оптимизировать сканирование table_1 на основе lookup_ids из table_2 и / или на основе разделов?

Вы можете попробовать отфильтровать ненужные записи. Эта концепция называется Spark predicate push down в Spark SQL-запросах, поэтому даже перед загрузкой данных в память Spark отфильтрует ненужные столбцы .. подробнее здесь

Передача предиката Spark в базу данных позволяет лучше оптимизировать запросы Spark. Предикат - это условие запроса, возвращающее истину или ложь, обычно находящееся в предложении WHERE. Смещение предиката вниз фильтрует данные в запросе к базе данных, уменьшая количество записей, извлекаемых из базы данных, и улучшая производительность запроса. По умолчанию Spark Dataset API автоматически отправляет действительные предложения WHERE в базу данных.