Python的 Polars脚本会导致内存泄漏并崩溃——它将CSV读取为惰性数据帧并写入数据库
我写了一个小脚本，使用Polars（Python版本）库扫描CSV，选择特定列、在lazyframe中筛选特定行，并将结果上传到Postgres数据库。
脚本在较小的测试CSV上可以工作，但是当我尝试在我的数据集中使用一个“真实”的CSV时，Python的进程会持续累积内存，直到崩溃。
我在Windows和 macOS上都做了测试，第一台计算机有64 GB内存，后者有32 GB，所以我很确定这不是硬件问题。
无论我在网上查了多少资料，似乎都没能解决我的问题。将low_memory设置为true、collect(engine='streaming') 或使用sink_csv都不起作用。
这是我的代码：
import polars as pl
from sqlalchemy import create_engine
result = (
pl.scan_csv("/path/to/file/201601.csv",
separator="|",
encoding='utf8-lossy',
try_parse_dates=True
)
#selecting specific columns
.select(
["ProgramClassID", "NetworkAffiliationID", "LogServiceID", "LogEntryDate","StartTime", "EndTime","Duration", "ProgramTitle", "Subtitle"]
)
#Filtering out all other broadcasters than SRC/3680
.filter(
pl.col("LogServiceID") != 3680)
#sorting by date
.sort(
"LogEntryDate")
#collecting to dataframe
.collect()
)
#defining login details for db
uri='postgresql://postgres:[email protected]:5432/crtc_program_logs'
#writing to db
result.write_database(table_name="polars_test_02",
connection=uri,
if_table_exists="replace"
)
这里 是我正在使用的CSV文件。
解决方案
对我而言，你的示例在CSV解析阶段就失败了。
我下载了 https://applications.crtc.gc.ca/OpenData/Television%20Logs/BroadcastLogs_2016_Q1_M1.csv（893 MB）
% ls -lh BroadcastLogs_2016_Q1_M1.csv
-rw-r--r--@ 1 user group 893M 2 Feb 2019 BroadcastLogs_2016_Q1_M1.csv
该文件没有任何行，在这种情况下的行结束符似乎就是 \r。
% wc -l BroadcastLogs_2016_Q1_M1.csv
0 BroadcastLogs_2016_Q1_M1.csv
Polars读取它后返回一个0 行、1.35亿列的数据帧。
>>> import polars as pl
>>> pl.read_csv("BroadcastLogs_2016_Q1_M1.csv", separator="|", try_parse_dates=True)
# shape: (0, 135_358_069)
# ┌────────────────┬──────────────┬─────────┬────────────┬───┬────────────────────────────────┬──────────────────────────────┬──────────────────────┬─────────────────────────────────┐
# │ BroadcastLogID ┆ LogServiceID ┆ LogDate ┆ SequenceNO ┆ … ┆ 03:51:40.0000000_duplicated_67 ┆ 2016-01-31_duplicated_140104 ┆ _duplicated_81004761 ┆ Match.com - "Jordan"|03:51:25.… │
# │ --- ┆ --- ┆ --- ┆ --- ┆ ┆ --- ┆ --- ┆ --- ┆ --- │
# │ str ┆ str ┆ str ┆ str ┆ ┆ str ┆ str ┆ str ┆ str │
# ╞════════════════╪══════════════╪═════════╪════════════╪═══╪════════════════════════════════╪══════════════════════════════╪══════════════════════╪═════════════════════════════════╡
# └────────────────┴──────────────┴─────────┴────────────┴───┴────────────────────────────────┴──────────────────────────────┴──────────────────────┴─────────────────────────────────┘
在我的系统上，耗时大约2 分钟，内存消耗约30GB。
Elapsed time: 110.63118 seconds
Memory usage: 28.9 GB
除了 \r 作为行结束符之外，文件似乎还使用 latin-1 编码，并且在 ProgramTitle 列中还包含了引号混用/破损的记录。
>>> import io
>>> import csv
>>> line1 = '830583776|3231|2016-01-01|17271||||1|||||20|||00:00:20.0000000||2016-01-16||"Design V.I.P.-#121-Herby Moreau-SH19 | Thalassa"|11:18:32.0000000|||||||||'
>>> line2 = '434824785|3456|2016-01-01|6|||||||||21|||00:00:25.0000000|08:30:28.0000000|2016-01-01||"9 EXAME INF 25""|08:30:03.0000000|||||||||'
>>>
>>> for line in [line1, line2]:
... print(len(next(csv.reader(io.StringIO(line), delimiter="|"))))
30
20
在这种情况下，第二行可以用 csv.QUOTE_NONE 解析以得到正确的列数，但随后第1 行就会断裂。
>>> for line in [line1, line2]:
... print(len(next(csv.reader(io.StringIO(line), delimiter="|", quoting=csv.QUOTE_NONE))))
31
30
对我而言，获取干净数据且不丢弃行/列的最简单方法是逐行“手动”解析：
num_cols = 30
with open("BroadcastLogs_2016_Q1_M1.csv", encoding="latin-1") as input_file:
with open("clean.csv", "w") as output_file:
writer = csv.writer(output_file)
for line in input_file:
row = next(csv.reader(io.StringIO(line), delimiter="|"))
if len(row) != num_cols:
row = next(csv.reader(io.StringIO(line), delimiter="|", quoting=csv.QUOTE_NONE))
assert len(row) == num_cols
writer.writerow(row)
其他任何一种一次性解析数据的方式都会因为引号问题把多行合并在一起。
于是我得到 6_499_105 行。
>>> from pathlib import Path
>>> Path("BroadcastLogs_2016_Q1_M1.csv").read_bytes().count(b"\r")
6499106
>>> pl.read_csv("clean.csv", try_parse_dates=True)
# shape: (6_499_105, 30)
# ┌────────────────┬──────────────┬────────────┬────────────┬───┬───────────┬───────────┬───────────┬───────────┐
# │ BroadcastLogID ┆ LogServiceID ┆ LogDate ┆ SequenceNO ┆ … ┆ Producer1 ┆ Producer2 ┆ Language1 ┆ Language2 │
# │ --- ┆ --- ┆ --- ┆ --- ┆ ┆ --- ┆ --- ┆ --- ┆ --- │
# │ i64 ┆ i64 ┆ date ┆ i64 ┆ ┆ str ┆ str ┆ str ┆ str │
# ╞════════════════╪══════════════╪════════════╪════════════╪═══╪═══════════╪═══════════╪═══════════╪═══════════╡
# │ 739036613 ┆ 3157 ┆ 2016-01-01 ┆ 1 ┆ … ┆ null ┆ null ┆ null ┆ null │
# │ 739036614 ┆ 3157 ┆ 2016-01-01 ┆ 2 ┆ … ┆ null ┆ null ┆ null ┆ null │
# │ 739036615 ┆ 3157 ┆ 2016-01-01 ┆ 3 ┆ … ┆ null ┆ null ┆ null ┆ null │
# │ 739036616 ┆ 3157 ┆ 2016-01-01 ┆ 4 ┆ … ┆ null ┆ null ┆ null ┆ null │
# │ 739036617 ┆ 3157 ┆ 2016-01-01 ┆ 5 ┆ … ┆ null ┆ null ┆ null ┆ null │
# │ … ┆ … ┆ … ┆ … ┆ … ┆ … ┆ … ┆ … ┆ … │
# │ 721758780 ┆ 3915 ┆ 2016-01-01 ┆ 3003 ┆ … ┆ null ┆ null ┆ null ┆ null │
# │ 721758781 ┆ 3915 ┆ 2016-01-01 ┆ 3004 ┆ … ┆ null ┆ null ┆ null ┆ null │
# │ 721758782 ┆ 3915 ┆ 2016-01-01 ┆ 3005 ┆ … ┆ null ┆ null ┆ null ┆ null │
# │ 721758783 ┆ 3915 ┆ 2016-01-01 ┆ 3006 ┆ … ┆ null ┆ null ┆ null ┆ null │
# │ 721758784 ┆ 3915 ┆ 2016-01-01 ┆ 3007 ┆ … ┆ null ┆ null ┆ null ┆ null │
# └────────────────┴──────────────┴────────────┴────────────┴───┴───────────┴───────────┴───────────┴───────────┘
# Elapsed time: 0.20965 seconds
# Memory usage: 3.7GB
与单次“读取”方法的对比
如果我们仅把数据转换为utf8，然后用Polars一次性全部解析：
>>> Path("utf8.csv").write_text(
... Path("BroadcastLogs_2016_Q1_M1.csv").read_text(encoding="latin-1", newline=""),
... newline=""
... )
# 936092544
>>> pl.read_csv("utf8.csv", separator="|", eol_char="\r", quote_char=None)
# ComputeError: found more fields than defined in 'Schema'
# Consider setting 'truncate_ragged_lines=True'.
这个错误意味着某些行中存在“额外的列”。
如果把它们截断，看起来就能得到完整的 6_499_105 行和 30 列：
>>> pl.read_csv("utf8.csv", separator="|", eol_char="\r", quote_char=None, truncate_ragged_lines=True)
# shape: (6_499_105, 30)
# ┌────────────────┬──────────────┬────────────┬────────────┬───┬───────────┬───────────┬───────────┬───────────┐
# │ BroadcastLogID ┆ LogServiceID ┆ LogDate ┆ SequenceNO ┆ … ┆ Producer1 ┆ Producer2 ┆ Language1 ┆ Language2 │
# │ --- ┆ --- ┆ --- ┆ --- ┆ ┆ --- ┆ --- ┆ --- ┆ --- │
# │ i64 ┆ i64 ┆ str ┆ i64 ┆ ┆ str ┆ str ┆ str ┆ str │
# ╞════════════════╪══════════════╪════════════╪════════════╪═══╪═══════════╪═══════════╪═══════════╪═══════════╡
# │ 739036613 ┆ 3157 ┆ 2016-01-01 ┆ 1 ┆ … ┆ null ┆ null ┆ null ┆ null │
# │ 739036614 ┆ 3157 ┆ 2016-01-01 ┆ 2 ┆ … ┆ null ┆ null ┆ null ┆ null │
# │ 739036615 ┆ 3157 ┆ 2016-01-01 ┆ 3 ┆ … ┆ null ┆ null ┆ null ┆ null │
# │ 739036616 ┆ 3157 ┆ 2016-01-01 ┆ 4 ┆ … ┆ null ┆ null ┆ null ┆ null │
# │ 739036617 ┆ 3157 ┆ 2016-01-01 ┆ 5 ┆ … ┆ null ┆ null ┆ null ┆ null │
# │ … ┆ … ┆ … ┆ … ┆ … ┆ … ┆ … ┆ … ┆ … │
# │ 721758780 ┆ 3915 ┆ 2016-01-01 ┆ 3003 ┆ … ┆ null ┆ null ┆ null ┆ null │
# │ 721758781 ┆ 3915 ┆ 2016-01-01 ┆ 3004 ┆ … ┆ null ┆ null ┆ null ┆ null │
# │ 721758782 ┆ 3915 ┆ 2016-01-01 ┆ 3005 ┆ … ┆ null ┆ null ┆ null ┆ null │
# │ 721758783 ┆ 3915 ┆ 2016-01-01 ┆ 3006 ┆ … ┆ null ┆ null ┆ null ┆ null │
# │ 721758784 ┆ 3915 ┆ 2016-01-01 ┆ 3007 ┆ … ┆ null ┆ null ┆ null ┆ null │
# └────────────────┴──────────────┴────────────┴────────────┴───┴───────────┴───────────┴───────────┴───────────┘
然而，在 try_parse_dates=True 时它会报错：
>>> pl.read_csv("utf8.csv", separator="|", eol_char="\r", quote_char=None, truncate_ragged_lines=True, try_parse_dates=True)
# InvalidOperationError: conversion from `str` to `time` failed in column 'StartTime' for 329 out of 23264 values: [" Bon"", " Bon"", … " Bon""]
这是因为 quote_char=None 会扭曲某些记录的 ProgramTitle 列，嵌入的 "| ..." 字符串最终出现在相邻的 StartTime 列中。
quote_char=None 本质上相当于Polars对应的 csv.QUOTE_NONE，因此你最终得到与数据不一致的问题，因为数据中既有需要启用它的行，也有需要禁用它的行以便正确解析。
站内所有文章版权归属LeftHeroAI导航站，无授权禁止任何主体转载、抄袭、复制内容，亦不得私自架设镜像站点。一经侵权，本站将通过法律途径追责。
相关文章
汇集全球数千款 AI 工具，每日持续更新，配套 AI 开发学习资源，轻松用 AI 提升工作效率。快捷键 Ctrl+D/⌘+D 一键收藏本站。