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

問題

標準方法，例如使用 pandas.read_csv()，在處理大量 CSV 檔案時通常會出現不足。這些方法是單線程的，由於磁碟 I/O 或記憶體限制，很快就會成為瓶頸。

---

終極 Python 程式設計師實作測驗

---

解決方案

透過並行化 CSV 操作，您可以利用多個 CPU 核心更快、更有效率地處理資料。本指南概述了使用以下技術：

1. Dask：對 pandas 程式碼進行最小變更的平行計算。
2. Polars：高效能 DataFrame 函式庫。
3. Python 的多處理模組：自訂並行化。
4. 檔案分割：使用較小的區塊進行分割和征服。

---

技巧

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)

關鍵考慮因素

1. 磁碟 I/O 與 CPU 限制
   
   確保您的並行策略平衡 CPU 處理與磁碟讀取/寫入速度。根據您的瓶頸是 I/O 還是計算進行最佳化。

2. 記憶體開銷
   
   與手動多重處理相比，Dask 或 Polars 等工具更節省記憶體。選擇符合您系統記憶體限制的工具。

3. 錯誤處理
   
   並行處理會帶來偵錯和錯誤管理的複雜性。實施強大的日誌記錄和異常處理以確保可靠性。

---

終極 Python 程式設計師實作測驗

以上是在 Python 中優化大規模資料處理：並行化 CSV 操作指南的詳細內容。更多資訊請關注PHP中文網其他相關文章！