在 Python 中優化大規模資料處理：並行化 CSV 操作指南-Python教學-PHP中文網

在 Python 中優化大規模資料處理：並行化 CSV 操作指南

DDD

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

原創

228 人瀏覽過

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)

登入後複製