在 Python 中优化大规模数据处理：并行化 CSV 操作指南-Python教程-PHP中文网

在 Python 中优化大规模数据处理：并行化 CSV 操作指南

DDD

发布： 2024-12-13 06:26:15

原创

232 人浏览过

Optimizing Large-Scale Data Processing in Python: A Guide to Parallelizing CSV Operations

问题

标准方法，例如使用 pandas.read_csv()，在处理大量 CSV 文件时通常会出现不足。这些方法是单线程的，由于磁盘 I/O 或内存限制，很快就会成为瓶颈。

终极 Python 程序员实践测试

解决方案

通过并行化 CSV 操作，您可以利用多个 CPU 核心更快、更高效地处理数据。本指南概述了使用以下技术：

Dask：对 pandas 代码进行最小更改的并行计算。
Polars：高性能 DataFrame 库。
Python 的多处理模块：自定义并行化。
文件分割：使用较小的块进行划分和征服。

技巧

1.分割大文件

将大型 CSV 文件分解为较小的块可以进行并行处理。这是一个示例脚本：

import os

def split_csv(file_path, lines_per_chunk=1000000):
    with open(file_path, 'r') as file:
        header = file.readline()
        file_count = 0
        output_file = None
        for i, line in enumerate(file):
            if i % lines_per_chunk == 0:
                if output_file:
                    output_file.close()
                file_count += 1
                output_file = open(f'chunk_{file_count}.csv', 'w')
                output_file.write(header)
            output_file.write(line)
        if output_file:
            output_file.close()
    print(f"Split into {file_count} files.")

登录后复制

2.使用 Dask 进行并行处理

Dask 是用 Python 处理大规模数据的游戏规则改变者。它可以毫不费力地并行化大型数据集上的操作：

import dask.dataframe as dd

# Load the dataset as a Dask DataFrame
df = dd.read_csv('large_file.csv')

# Perform parallel operations
result = df[df['column_name'] > 100].groupby('another_column').mean()

# Save the result
result.to_csv('output_*.csv', single_file=True)

登录后复制

Dask 通过对数据块进行操作并在可用内核之间智能地调度任务来处理内存限制。

终极 Python 程序员实践测试

3.用 Polar 来增压

Polars 是一个相对较新的库，它将 Rust 的速度与 Python 的灵活性结合在一起。它是为现代硬件设计的，处理 CSV 文件的速度比 pandas 快得多：

import polars as pl

# Read CSV using Polars
df = pl.read_csv('large_file.csv')

# Filter and aggregate data
filtered_df = df.filter(pl.col('column_name') > 100).groupby('another_column').mean()

# Write to CSV
filtered_df.write_csv('output.csv')

登录后复制

Polars 在速度和并行性至关重要的情况下表现出色。它对于多核系统特别有效。

4.多处理手动并行

如果您希望控制处理逻辑，Python 的多处理模块提供了一种并行化 CSV 操作的简单方法：

from multiprocessing import Pool
import pandas as pd

def process_chunk(file_path):
    df = pd.read_csv(file_path)
    # Perform operations
    filtered_df = df[df['column_name'] > 100]
    return filtered_df

if __name__ == '__main__':
    chunk_files = [f'chunk_{i}.csv' for i in range(1, 6)]
    with Pool(processes=4) as pool:
        results = pool.map(process_chunk, chunk_files)

    # Combine results
    combined_df = pd.concat(results)
    combined_df.to_csv('final_output.csv', index=False)

登录后复制