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的资料请关注其它相关文章！