Python中如何将30M的parquet文件转为pandas dataframe并存入MySQL，却耗尽了6G内存？

python 代码

from pyarrow import parquet as pq

fs = s3fs.S3FileSystem(…)


path = [‘s3a://…’]


r = pq.ParquetDataset(_path, filesystem=fs)
df = r.read().to_pandas()
df.to_sql(…)

相关的日志

Jan 18 14:18:33  kernel: Out of memory: Kill process 3342 (python3) score 842 or sacrifice child
Jan 18 14:18:33  kernel: Killed process 3342 (python3) total-vm:7950428kB, anon-rss:6945048kB, file-rss:0kB, shmem-rss:0kB

Python中如何将30M的parquet文件转为pandas dataframe并存入MySQL，却耗尽了6G内存？

2 回复

这问题我遇到过，parquet文件看着不大但内存爆了，通常是这几个原因：

  1. 数据类型膨胀：parquet存储时很紧凑，但读到pandas时类型可能不匹配，比如int32变成int64，内存直接翻倍
  2. 字符串处理：object类型在pandas里是Python对象，每个值都单独存，特别吃内存
  3. 索引开销：pandas的DataFrame索引和列名都有额外开销

试试这个优化方案：

import pandas as pd
import pyarrow.parquet as pq
from sqlalchemy import create_engine
import gc

def parquet_to_mysql(file_path, table_name, chunksize=100000):
    # 1. 用pyarrow直接读取，避免pandas的额外转换
    table = pq.read_table(file_path)
    
    # 2. 指定数据类型，防止自动类型推断膨胀
    df = table.to_pandas(types_mapper={
        pd.StringDtype(): "string",  # 使用pandas的string类型，比object省内存
        pd.Int64Dtype(): "int64",
        pd.Float64Dtype(): "float64"
    })
    
    # 3. 删除原始table释放内存
    del table
    gc.collect()
    
    # 4. 分块写入MySQL
    engine = create_engine('mysql+pymysql://user:password@localhost/dbname')
    
    # 先写入表结构
    df.head(0).to_sql(table_name, engine, if_exists='replace', index=False)
    
    # 分块写入数据
    for i in range(0, len(df), chunksize):
        chunk = df.iloc[i:i+chunksize]
        chunk.to_sql(table_name, engine, if_exists='append', index=False)
        print(f"写入 {i+len(chunk)}/{len(df)} 行")
        
        # 及时释放内存
        del chunk
        gc.collect()
    
    return len(df)

# 使用示例
file_path = "your_file.parquet"
row_count = parquet_to_mysql(file_path, "your_table")
print(f"总共写入 {row_count} 行数据")

关键点：

  • 用pyarrow直接读比pd.read_parquet更省内存
  • 显式指定数据类型，特别是把object列转为string类型
  • 分块写入，避免一次性加载所有数据到内存

如果还不行，考虑用Dask处理大数据，或者直接检查parquet文件的实际数据量。

建议：检查数据类型并分块处理。


漏了点处理
df[’_time’] = df[‘time’].apply(lambda x: parse(x[:11] + " " + x[12:]).strftime("%Y-%m-%d %H:%M:%S"))
df[‘bytes’] = df[‘request_size’] + df[‘response_size’]
df[‘bytes’] = df[‘bytes’].astype(int)

回到顶部