Большой объем разбора URL-адресов в Python ⇐ Python

[Anonymous]

Я использую пакеты Polars, urllib и tldextract в Python для анализа двух столбцов строк URL-адресов в файлах паркета, сжатых с помощью zstd (в среднем 8 ГБ, 40 миллионов строк). Проанализированные выходные данные включают схему, netloc, поддомен, домен, суффикс, путь, запрос и фрагмент. На моем ПК обработка одного файла занимает около 16 минут (64 ГБ ОЗУ, 32 логических ядра). Входные данные считываются с одного SSD (x:/), а выходные данные записываются на отдельный SSD (y:/).
Мой код не использует многопроцессорность. Для обеспечения эффективности он опирается на функции потоковой передачи и векторизации поляров. Использование оперативной памяти при обработке близко к пределу. Я не могу в полной мере воспользоваться преимуществами векторизации поляров, поскольку urllib и tldextract не являются встроенными в движок поляров (rust).
Существуют ли альтернативные подходы или модификации, которые помогут ускорить обработку? Разумно ли время обработки в 16 минут? Я подумывал о создании расширения Rust для поляров, которое имеет ту же функциональность, что и urllib и tldextract, но это похоже на изобретение велосипеда.
Метод ниже выполняет тяжелую работу (python 3.12, поляры 1.31)
def build_parsed_url_for_date(
event_date: str,
silver_root: Path,
gold_root: Path,
compression: str = "zstd",
) -> None:
"""
For a given event_date (YYYY-MM-DD):

- Read SILVER parquet from: silver_root / f"event_date={event_date}" / *.parquet
- Parse `url` and `referrer` columns into components (PSL-based).
- Write GOLD/parsed_url parquet to:
gold_root / "parsed_url" / f"event_date={event_date}" / "part.parquet"
"""
# input paths
silver_partition = silver_root / f"event_date={event_date}"
if not silver_partition.exists():
raise FileNotFoundError(f"SILVER partition not found: {silver_partition}")

silver_files = sorted(silver_partition.glob("*.parquet"))
if not silver_files:
raise FileNotFoundError(f"No parquet files in {silver_partition}")

# polars lazy frame
lf = pl.scan_parquet([str(f) for f in silver_files])

# wrappers for urllib and tldextract parsers
parser_url = make_polars_parser("url") # returns dict of parsed components
parser_ref = make_polars_parser("ref")

# Predicate pushdown
lf_parsed = (
lf.select(["id", "referrer", "url"])
.with_columns(
pl.col("url").map_elements(parser_url, return_dtype=url_struct_dtype("url")).alias("url_parsed"),
pl.col("referrer").map_elements(parser_ref, return_dtype=url_struct_dtype("ref")).alias("ref_parsed"),
)
.unnest("url_parsed")
.unnest("ref_parsed")
)

# Output path
gold_parsed_root = gold_root / "parsed_url"
out_dir = gold_parsed_root / f"event_date={event_date}"
out_dir.mkdir(parents=True, exist_ok=True)
out_path = out_dir / "part.parquet"

# Stream/write to parquet
lf_parsed.sink_parquet(str(out_path), compression=compression)


Подробнее здесь: https://stackoverflow.com/questions/798 ... -in-python