Python处理海量复杂CSV数据：多进程与算法优化实战

"""鉴于项目保密需要，以下代码仅为示例"""
import time
import pathlib as pl
import pandas as pd
from threading import Thread
from multiprocessing import Queue, Process, cpu_count
# 导入多线程Thread,多进程的队列Queue,多进程Process，CPU核数cpu_count
# 存放分段读取的数据队列，注：maxsize控制队列的最大数量，避免一次性读取到内存中的数据量太大
data_queue = Queue(maxsize=cpu_count() * 2)  
# 存放等待写入磁盘的数据队列
write_queue = Queue()  
def read_data(path: pl.Path, data_queue: Queue, size: int = 10000):
    """
    读取数据放入队列的方法
    :return:
    """
    data_obj = pd.read_csv(path, sep=',', header=0, chunksize=size, dtype='category')
    for idx, df in enumerate(data_obj):
        while data_queue.full():  # 如果队列满了，那就等待
            time.sleep(1)
        data_queue.put((idx + 1, df))
    data_queue.put((None, None))  # 放入结束信号
def write_data(out_path: pl.Path, write_queue: Queue):
    """
    将数据增量写入CSV的方法
    :return:
    """
    while True:
        while write_queue.empty():
            time.sleep(1)
        idx, df = write_queue.get()
        if df is None:
            return  # 结束退出
        df.to_csv(out_path, mode='a', header=None, index=False, encoding='ansi')  # 输出CSV
def parse_data(data_queue: Queue, write_queue: Queue):
    """
    从队列中取出数据，并加工的方法
    :return:
    """
    while True:
        while write_queue.empty():
            time.sleep(1)
        idx, df = data_queue.get()
        if df is None:  # 如果是空的结束信号，则结束退出进程，
        # 特别注意结束前把结束信号放回队列，以便其他进程也能接收到结束信号！！！
            data_queue.put((idx, df))
            return
        """处理数据的业务逻辑略过"""
        write_queue.put((idx, df))  # 将处理后的数据放入写队列
# 创建一个读取数据的线程
read_pool = Thread(target=read_data, args=(read_data_queue, *args))
read_pool.start()  # 开启读取线程
# 创建一个增量写入CSV数据的线程
write_pool = Thread(target=write_data, args=(write_data_queue, *args))
write_pool.start()  # 开启写进程
pools = []  # 存放解析进程的队列
for i in range(cpu_count()):  # 循环开启多进程，不确定开多少个进程合适的情况下，那么按CPU的核数开比较合理
    pool = Process(target=parse_data, args=(read_data_queue, write_data_queue, *args))
    pool.start()  # 启动进程
    pools.append(pool)  # 加入队列
for pool in pools:
    pool.join()  # 等待所有解析进程完成
# 所有解析进程完成后，在写队列放入结束写线程的信号
write_data_queue.put((None, None))  
write_pool.join()  # 等待写线程结束
print('任务完成')

"""鉴于保密需要，以下代码仅为示例"""
class Obj:
    def __init__(self, ser: pd.Series):
        """
        初始化类对象
        :param ser: 传入series
        """
        self.ser = ser  # 行数据
        self.attrs1 = []  # 属性1
        self.attrs2 = []  # 属性2
        self.attrs3 = []  # 属性3
    def __repr__(self):
        """
        自定义输出
        """
        attrs1 = '_'.join([str(a) for a in self.attrs1])
        attrs2 = '_'.join([str(a) for a in self.attrs2])
        attrs3 = '_'.join([str(a) for a in self.attrs3])
        return '_'.join([attrs1, attrs2, attrs3])
    def run(self):
        """运行业务逻辑"""
# 创建obj列，存入类对象
data['obj'] = data.apply(lambda x: Obj(x), axis=1)
# 运行obj列中的类方法获得判断结果
data['obj'] = data['obj'].map(lambda x: x.run())
# 链式调用，1将类对象文本化->2拆分到多列->3删除空列->4转换为category格式
data[['col1', 'col2', 'col3', ...省略]] = data['obj'].map(str).str.split('_', expand=True).dropna(axis=1).astype('category')
# 删除obj列
data.drop(columns='obj', inplace=True)  

"""numpy方法使用示例"""
import numpy as np
# 计算数字的个数组合bincount
np.bincount([9, 2, 13, 12, 9, 10, 11])
# 输出结果：array([0, 0, 1, 0, 0, 0, 0, 0, 0, 2, 1, 1, 1, 1], dtype=int64)
# 取得个数最多的数字argmax
np.argmax(np.bincount([9, 2, 13, 12, 9, 10, 11]))
# 输出结果: 9
# 将数字按照个数优先，其次大小进行排序argsort
np.argsort(np.bincount([9, 2, 13, 12, 9, 10, 11]))
# 输出结果：array([ 0,  1,  3,  4,  5,  6,  7,  8,  2, 10, 11, 12, 13,  9], dtype=int64)
# 翻转列表flipud
np.flipud(np.argsort(np.bincount([9, 2, 13, 12, 9, 10, 11])))
# 输出结果: array([ 9, 13, 12, 11, 10,  2,  8,  7,  6,  5,  4,  3,  1,  0], dtype=int64)
# 查找相同值in1d
np.in1d([2, 3, 4], [2, 9, 3])
# 输出结果: array([ True,  True, False]) 注：指2,3True，4False
np.all(np.in1d([2, 3], [2, 9, 3]))
# 输出结果: array([ True,  True])
# 是否全是all
np.all(np.in1d([2, 3, 4], [2, 9, 3]))  # 判断组合1是否包含在组合2中
# 输出结果: False
np.all(np.in1d([2, 3], [2, 9, 3]))
# 输出结果: True