Python pola.rs OOM в файле размером 30 ГБ ⇐ Python

[Anonymous]

Мне нужно агрегировать необработанные данные временных рядов в OHLC с точностью до 1 м. Я создал код с помощью python-polar, который возвращает этот план:
SELECT [col("parsed_timestamp"), col("label1"), col("open"), col("high"), col("low"), col("close")] FROM
AGGREGATE
[
col("price").first().alias("open"),
col("price").max().alias("high"),
col("price").min().alias("low"),
col("price").last().alias("close")
] BY [col("label1")]
FROM
WITH_COLUMNS:
[col("timestamp").strict_cast(Datetime(Microseconds, None)).alias("parsed_timestamp")]
Csv SCAN [ticker.csv]
PROJECT */11 COLUMNS

Я использую scan_csv и Collect(streaming=True). Кроме того, я вызываю .group_by_dynamic("parsed_timestamp", Every="1m", group_by=["label1"]) (который по какой-то причине не отображается в плане).
Однако этот код по-прежнему использует 30 ГБ памяти при агрегировании файла размером 30 ГБ. Есть ли какие-либо оптимизации, которые я могу сделать? Данные сортируются по временным меткам в порядке возрастания.
Я создал простую программу, которая просто перебирает каждую строку, обновляет свечи в карте памяти, а затем выгружает их после завершения в стандартный вывод/файл. . Он вообще не использует никакой дополнительной памяти. Я полагаю, что-то подобное должно быть возможно и с полярами?
Код UPD:
input_file = "input.csv"
output_file = "output.csv"
query = (
pl.scan_csv(
input_file,
has_header=False,
new_columns=[
"label1",
"timestamp",
"price",
],
)
.with_columns(
pl.col("timestamp").cast(pl.Datetime).alias("parsed_timestamp"),
)
.group_by_dynamic("parsed_timestamp", every="1m", group_by=["label1"])
.agg(
pl.col("price").first().alias("open"),
pl.col("price").max().alias("high"),
pl.col("price").min().alias("low"),
pl.col("price").last().alias("close"),
)
.select(
[
"parsed_timestamp",
"label1",
"open",
"high",
"low",
"close",
]
)
)
query.collect(streaming=True).write_csv(output_file)


Подробнее здесь: https://stackoverflow.com/questions/787 ... -30gb-file