Помогает ли секционирование при чтении ключевых столбцов с помощью функции?
У меня есть этот фрейм данных:
+------+
|my_col|
+------+
|202101|
|202209|
+------+
При записи в виде файла паркета я разделяю его на основе столбца «my_col», поэтому у меня должно получиться два раздела (два файла паркета).
Затем я буду читать сохраненный набор данных, применяя фильтр.
- Я правильно понимаю, что Spark не будет загружать данные в память из файла, содержащего my_col=202101, при применении фильтра
.filter("my_col >= 202201")? - Будут ли данные из файла паркета, содержащего my_col=202101, загружаться в память при применении фильтра
.filter("substring(my_col, 1, 4) >= 2022")?
В последнем случае я не фильтрую напрямую значения ключевого столбца, а вместо этого к столбцу применяется функция. Интересно, поможет ли в этом случае разбиение на разделы сэкономить время чтения.
не заслуживает отдельного ответа (Ахмеду), но это верно только для обрезки разделов, где искра эффективно выполняет его, когда он отключен на уровне поля (предикаты нажатия/PushedFilters) разрешено только простое равенство /in, функции - нет . Это привело к пользовательской оптимизации качества, чтобы избежать «сканирования таблиц».
В зависимости от версии Spark и хранилища метаданных вы можете увидеть разницу в том, какие предикаты передаются для оценки в хранилище метаданных, и то, что Spark должен фильтровать на стороне клиента. См. Issues.apache.org/jira/plugins/servlet/mobile#issue/SPARK-33537
Ответы 1
Поэтому я попробовал приведенный ниже пример, чтобы выяснить, имеет ли какое-либо значение функция, используемая в фильтре. Пример:
data = [
("A ", "202201"),
("B ", "202209"),
("C ", "202210"),
("D ", "202301"),
]
columns = ["name", "dob"]
df = spark.createDataFrame(data, columns)
df.write.partitionBy("dob").mode("overwrite").parquet("people.parquet")
people = spark.read.parquet("people.parquet")
people_filtered_without_function = people.filter("dob >= 202201")
people_filtered_with_function = people.filter("substring(dob, 1, 4) >= 2022")
people_filtered_without_function.explain()
people_filtered_with_function.explain()
Вот результаты физических планов:
- Без функции
== Физический план == *(1) СтолбецToRow +- FileScan паркет [имя#7,доб#8] Пакетный: true, DataFilters: [], Формат: Паркет, Расположение: InMemoryFileIndex(1 путь)[файл:/opt/spark/work-dir/people.parquet], PartitionFilters: [isnotnull(dob#8), (dob#8 >= 202201)], PushedFilters: [], ReadSchema: structname:string
- С функцией
== Физический план == *(1) СтолбецToRow +- FileScan паркет [имя#7,доб#8] Пакетный: true, DataFilters: [], Формат: Паркет, Расположение: InMemoryFileIndex(1 путь)[файл:/opt/spark/work-dir/people.parquet], PartitionFilters: [isnotnull(dob#8), (cast(substring(cast(dob#8 as string), 1, 4) as int) >= 2022)], PushedFilters: [], ReadSchema: structname:string
Обратите внимание, что основное отличие заключается в следующем:
Фильтр разделов (без):
(доб#8 >= 202201)
Фильтр разделов (с):
(cast(substring(cast(dob#8 as string), 1, 4) as int) >= 2022)
Как вы можете видеть выше, разницу между этими двумя методами. Я бы сказал, что происходит вот что:
- Да, действительно данные будут фильтроваться спарком и считываться только соответствующие файлы.
- План также отражает использование функций в фильтре. Поэтому я бы предположил, что основным отличием в производительности здесь являются только накладные расходы на приведение типов и применение функций, а не фактическая логика фильтрации.
Хорошо, я не знал о PartitionFilters внутренних планах запросов. Спасибо! Похоже, что разделение по столбцу dob стоит того, даже если последующие фильтры чтения используют (базовые) функции.
Ты можешь объяснить