您好，登录后才能下订单哦！
密码登录
登录注册
点击 登录注册 即表示同意《亿速云用户服务条款》
这篇“Python如何处理复杂CSV文件”文章的知识点大部分人都不太理解，所以小编给大家总结了以下内容，内容详细，步骤清晰，具有一定的借鉴价值，希望大家阅读完这篇文章能有所收获，下面我们一起来看看这篇“Python如何处理复杂CSV文件”文章吧。
鉴于项目保密的需要，不便透露太多项目的信息，因此，简单介绍一下项目存在的难点：
海量数据：项目是对CSV文件中的数据进行处理，而特点是数据量大...真的大！！！拿到的第一个CSV示例文件是110多万行(小CASE)，而第二个文件就到了4500万行，等到第三个文件......好吧，一直没见到第三个完整示例文件，因为太大了，据说是第二个示例文件的40多倍，大概二十亿行......
业务逻辑复杂：项目是需要对CSV文件的每一行数据的各种组合可能性进行判断，而判断的业务逻辑较为复杂，如何在解决复杂逻辑的同时保证较高的处理效率是难点之一。
因为数据量太大，肯定是要分批对数据进行处理，否则，效率低不谈，大概率也没有足够的内存能够支撑，需要用到chunksize，此外，为了节约内存，以及提高处理效率，可以将文本类的数据存储为“category”格式：
项目整体是计算密集型的任务，因此，需要用到多进程，充分利用CPU的多核性能；
多线程进行读取与写入，其中，写入使用to_csv的增量写入方法，mode参数设置为'a'；
多进程与多线程开启一般为死循环，需要在合适的位置，放入结束循环的信号，以便处理完毕后退出多进程或多线程
"""鉴于项目保密需要，以下代码仅为示例"""
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('任务完成')将类对象存入dataframe列
在尝试了n种方案之后，最终使用了将类对象存到dataframe的列中，使用map方法，运行类方法，最后，将运行结果展开到多列中的方式。该方案本项目中取得了最佳的处理效率。
"""鉴于保密需要，以下代码仅为示例"""
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)减少计算次数以提高运行效率
在整个优化过程中，对运行效率产生最大优化效果的有两项：
一是改变遍历算法，采用直接对整行数据进行综合判断的方法，使原需要遍历22个组合的计算与判断大大减少
二是提前计算特征组合，制作成字典，后续直接查询结果，而不再进行重复计算
使用numpy加速计算
numpy还是数据处理上的神器，使用numpy的方法，比自己实现的方法效率要高非常多，本项目中就用到了：bincount、argsort，argmax、flipud、in1d、all等，即提高了运行效率，又解决了逻辑判断的问题：
"""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
优化前后的效率对比
以上就是关于“Python如何处理复杂CSV文件”这篇文章的内容，相信大家都有了一定的了解，希望小编分享的内容对大家有帮助，若想了解更多相关的知识内容，请关注亿速云行业资讯频道。
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。