python简简单单写个批量CSV合并工具(多进程)

1、功能快速实现批量CSV文件的合并，1000个CSV文件只需13秒合并完2、代码实例(多进程)import osimport csv
import multiprocessing
import time


def process_csv(file):
    filepath = os.path.join(r'D:\py\data', file)
    with open(filepath, 'r') as f:
        reader = csv.reader(f)

        rows = []
        for row in reader:
            # 处理CSV文件的每一行数据
            processed_row = process_row(row)
            rows.append(processed_row)

    return rows


def process_row(row):
    # 这里可以对每一行数据进行处理，根据实际需求进行操作
    processed_row = row
    # 示例：将每个元素转换为大写
    processed_row = [item.upper() for item in processed_row]
    return processed_row

def do(csv_files):
    pool = multiprocessing.Pool()
    results = pool.map(process_csv, csv_files)
    pool.close()
    pool.join()
 # 合并所有结果到一个列表
    merged_rows = []
    for rows in results:
        merged_rows.extend(rows)

    # 将合并后的结果写入到输出文件
    output_filepath = os.path.join(r'D:\py', 'output.csv')
    with open(output_filepath, 'w', newline='') as output_file:
        writer = csv.writer(output_file)
        writer.writerows(merged_rows)




if __name__ == '__main__':
    csv_files = os.listdir(r'D:\py\data')
    start_time = time.time()  # 记录起始时间

    do(csv_files)

    end_time = time.time()  # 记录结束时间
    elapsed_time = end_time - start_time  # 计算经过的时间
    print(f"执行耗时: {elapsed_time}秒")
3、效果图

程序启动时CPU使用峰值到达100%。

4、实现过程讲解

在这段代码中，pool.map(process_csv, csv_files)的作用是使用多进程方式并行处理多个CSV文件。具体解释如下：

1. pool是一个multiprocessing.Pool()对象，创建了一个进程池。
2. pool.map(process_csv, csv_files)调用map()方法，该方法接收两个参数：process_csv函数和csv_files列表。map()方法会将csv_files列表中的每个元素（即CSV文件名）传递给process_csv函数，并以多进程方式并行执行process_csv函数。
3. 在process_csv函数中，打开指定的CSV文件，并逐行读取文件内容，对每一行数据调用process_row函数进行处理。
4. process_row函数是一个示例函数，用于对每一行数据进行处理。在这里，它将每个元素都转换为大写形式。
5. 处理完成后，将每个CSV文件的处理结果存储到results列表中。

5、用单进程读取同样的文件效率对比

1、下面用单进程方式合并同样数据的文件则需要耗时20秒，比多进程方式慢了将近一半

2、单进程运行效果图

3、代码实例(单进程)import osimport csv
from timer import get_time


def process_csv(file):
    filepath = os.path.join(r'D:\py\data', file)
    with open(filepath, 'r') as f:
        reader = csv.reader(f)

        rows = []
        for row in reader:
            # 处理CSV文件的每一行数据
            processed_row = process_row(row)
            rows.append(processed_row)

    return rows


def process_row(row):
    # 这里可以对每一行数据进行处理，根据实际需求进行操作
    processed_row = row
    # 示例：将每个元素转换为大写
    processed_row = [item.upper() for item in processed_row]
    return processed_row


@get_time
def do(csv_files):
    results = []
    for file in csv_files:
        result = process_csv(file)
        results.append(result)
    merged_rows = []
    for rows in results:
        merged_rows.extend(rows)
    #将合并后的结果写入到输出文件
    output_filepath = os.path.join(r'D:\py', 'output.csv')
    with open(output_filepath, 'w', newline='') as output_file:
        writer = csv.writer(output_file)
        writer.writerows(merged_rows)

if __name__ == '__main__':
    csv_files = os.listdir(r'D:\py\data')
    do(csv_files)

6、总结

当我们使用多进程编程时，我们可以同时运行多个进程，每个进程都有自己的一份代码、数据和资源。Python提供了多个库来实现多进程编程，其中最常用的是multiprocessing模块。

Python中的多进程编程有以下几个优点：

1. 提高计算性能：通过并行执行多个进程，可以充分利用多核处理器的计算能力，加快程序的运行速度。
2. 实现异步任务：多进程可以同时执行多个任务，其中一个任务阻塞不会影响其他任务的执行，从而实现异步任务的效果。
3. 充分利用资源：多进程可以分配和管理系统的资源，使得程序可以更好地利用计算机的硬件资源。

下面是Python多进程编程的一般步骤：

1. 导入multiprocessing模块：首先需要导入multiprocessing模块，它提供了创建和管理进程的类和函数。
2. 创建进程：使用multiprocessing.Process()类可以创建一个新的进程。
3. 定义进程要执行的任务：将要在新进程中执行的代码放入一个函数或方法中。
4. 启动进程：调用进程对象的start()方法启动新进程。
5. 等待进程结束：可以使用join()方法等待进程执行完成。