Python中单机处理上亿条日志（每条10+列）的高效方法？

对于单机处理上亿条日志这种量级的数据，核心思路是 “分而治之” + “惰性处理”，避免一次性将数据全读进内存。直接上干货，一个基于 pandas 的 chunksize 和 dask 的实用方案：

import pandas as pd
import dask.dataframe as dd
from dask.diagnostics import ProgressBar
import os

# 方法1：使用pandas的chunksize进行分块处理（适合复杂逐行逻辑）
def process_with_pandas_chunks(file_path, output_path, chunk_size=100000):
    """
    分块读取CSV日志文件，处理每个块，并追加到输出文件。
    这是最经典、控制粒度最细的方法。
    """
    # 第一次读取获取列名
    first_chunk = pd.read_csv(file_path, nrows=1)
    columns = first_chunk.columns
    
    # 初始化输出文件（写入列名）
    first_chunk.to_csv(output_path, index=False)
    
    # 从第二行开始分块读取和处理
    chunk_iter = pd.read_csv(file_path, chunksize=chunk_size, skiprows=1, names=columns)
    
    for i, chunk in enumerate(chunk_iter):
        # 在这里进行你的实际数据处理，例如：
        # 1. 过滤：chunk = chunk[chunk['status'] == 'ERROR']
        # 2. 计算：chunk['new_col'] = chunk['col1'] + chunk['col2']
        # 3. 聚合等...
        
        # 示例：简单过滤并添加时间戳转换
        chunk['timestamp'] = pd.to_datetime(chunk['timestamp'], errors='coerce')
        chunk = chunk.dropna(subset=['timestamp'])
        
        # 追加到输出文件，模式为'a'（追加），不写header
        chunk.to_csv(output_path, mode='a', header=False, index=False)
        
        print(f"已处理第 {i+1} 块，共 {len(chunk)} 行")

# 方法2：使用Dask进行并行和惰性计算（适合列式分析和聚合）
def process_with_dask(file_path, output_path):
    """
    使用Dask创建惰性DataFrame，进行分布式计算，适合大规模聚合和转换。
    """
    # 创建Dask DataFrame（数据仍在磁盘，未加载到内存）
    ddf = dd.read_csv(file_path, assume_missing=True)  # assume_missing有助于推断数据类型
    
    # 进行惰性转换操作，例如：
    # 1. 类型转换
    ddf['timestamp'] = dd.to_datetime(ddf['timestamp'], errors='coerce')
    # 2. 过滤
    ddf = ddf.dropna(subset=['timestamp'])
    # 3. 聚合示例（按小时统计）
    result = ddf.groupby(ddf['timestamp'].dt.hour).size().compute()
    
    # 如果要输出处理后的完整数据到单个CSV（可能会拆分成多个分区文件）
    ddf.to_csv(output_path, single_file=False, index=False)
    
    # 显示进度
    with ProgressBar():
        result = ddf.compute()  # 实际触发计算
    return result

# 根据场景选择执行
if __name__ == "__main__":
    log_file = "your_large_log.csv"  # 你的日志文件路径
    output_file = "processed_log.csv"
    
    # 方案1：更细粒度控制，适合行级复杂操作
    process_with_pandas_chunks(log_file, output_file, chunk_size=50000)
    
    # 方案2：更高效的并行聚合，适合分析任务
    # process_with_dask(log_file, "dask_output")

关键点解析：

1. pandas的chunksize：通过迭代器逐块处理，内存占用恒定（约等于一个块的大小）。适合需要逐行判断、跨行逻辑不强的场景。你可以把处理逻辑写在for chunk in chunk_iter:循环里。

2. Dask：它在后台将数据自动分割成多个分区，并行处理。语法和pandas很像，但操作是惰性的，只有调用.compute()时才真正执行。特别适合做分组、聚合、筛选等列式运算。对于上亿条数据，它能充分利用多核。

3. 文件格式：如果日志是纯文本且字段固定，也可以考虑先按行切分成多个小文件，再用multiprocessing并行处理每个文件，最后合并结果。对于CSV/JSON，上述方法更直接。

4. 列存储：如果后期需要频繁分析部分列，考虑将数据转换为Parquet格式。它按列存储，查询时只读取需要的列，速度极快。用ddf.to_parquet("output.parquet")即可。

5. 数据类型：在读取时指定dtype或使用low_memory=False可以避免pandas的内存警告和类型误判。

一句话建议：先用pandas的chunksize快速验证处理逻辑，再用Dask或转Parquet进行全量处理。

spark+1