python3线程池ThreadPoolExecutor处理csv文件数据
目录
- 背景
- 知识点
- 拓展
- 库
- 流程
- 实现代码
- 解释
背景
由于不同乙方对服务商业务接口字段理解不一致,导致线上上千万数据量数据存在问题,为了修复数据,通过 Python 脚本进行修改
知识点
Python3、线程池、pymysql、CSV 文件操作、requests
拓展
当我们程序在使用到线程、进程或协程的时候,以下三个知识点可以先做个基本认知
CPU 密集型、IO 密集型、GIL 全局解释器锁
库
pip3 install requests
pip3 install pymysql
流程
实现代码
# -*- coding:utf-8 -*- # @FileName:grade_update.py # @Desc :在一台超级计算机上运行过的牛逼Python代码 import time from concurrent.futures import ThreadPoolExecutor,FIRST_COMPLETED,wait import requests import pymysql from projectPath import path gradeId = [4303, 4304, 1000926, 1000927] def writ_mysql(): """ 数据库连接 """ return pymysql.connect(host="localhost", port=3306, user="admin", password="admin", database="test" ) def oprationdb(grade_id, member_id): """ 操作数据库 """ db = writ_mysql() try: cursor = db.cursor() sql = f"UPDATE `t_m_member_grade` SET `current_grade_id`={grade_id}, `modified` =now() WHERE `member_id`={member_id};" cursor.execute(sql) db.commit() print(f"提交的SQL->{sql}") except pymysql.Error as e: db.rollback() print("DB数据库异常:", e) db.close() return True def interface(rows, thead): """ 调用第三方接口 """ print(f"处理数据行数--->{thead}----数据--->{rows}") try: url = "http://xxxx/api/xxx-data/Tmall/bindQuery" body = { "nickname": str(rows[0]), "seller_name": "test", "mobile": "111" } heade={"Content-Type": "application/x-www-form-urlencoded"} res = requests.post(url=url, data=body,headers=heade) result = res.json() if result["data"]["status"] in [1, 2]: grade = result["data"]["member"]["level"] grade_id = gradeId[grade] oprationdb(grade_id=grade_id, member_id=rows[1]) return True return True except Exception as e: print(f"调用异常:{e}") def read_csv(): import csv # db = writ_mysql() #线程数 MAX_WORKERS=5 with ThreadPoolExecutor(MAX_WORKERS) as pool: with open(path + '/file/result2_colu.csv', 'r', newline='', encoding='utf-8') as f: #set() 函数创建无序不重复元素集 seq_notdone = set() seq_done = set() # 使用csv的reader()方法,创建一个reader对象 reader = csv.reader(f) n = 0 for row in reader: n += 1 # 遍历reader对象的每一行 try: seq_notdone.add(pool.submit(interface, rows=row, thead=n)) if len(seq_notdone) >= MAX_WORKERS: #FIRST_COMPLETED文档说明 -- Return when any future finishes or is cancelled. done, seq_notdone = wait(seq_notdone,return_when=FIRST_COMPLETED) seq_done.update(done) except Exception as e: print(f"解析结果出错:{e}") # db.close() return "完成" if __name__ == '__main__': read_csv()
解释
引入线程池库
from concurrent.futures import ThreadPoolExecutor,FIRST_COMPLETED,wait
pool.submit(interface, rows=row, thead=n)
提交任务,interface 调用的函数,rows、thead 为 interface() 函数的入参
任务持续提交,线程池通过 MAX_WORKERS 定义的线程数持续消费
说明像这种 I/O 密集型的操作脚本适合使用多线程,如果是 CPU 密集型建议使用进行,根据机器核数进行配置
以上就是python3线程池ThreadPoolExecutor处理csv文件数据的详细内容,更多关于python3 ThreadPoolExecutor处理csv的资料请关注我们其它相关文章!
赞 (0)