怎么使用python3線程池ThreadPoolExecutor處理csv文件數(shù)據(jù)

這篇文章主要介紹“怎么使用python3線程池ThreadPoolExecutor處理csv文件數(shù)據(jù)”的相關(guān)知識(shí)，小編通過(guò)實(shí)際案例向大家展示操作過(guò)程，操作方法簡(jiǎn)單快捷，實(shí)用性強(qiáng)，希望這篇“怎么使用python3線程池ThreadPoolExecutor處理csv文件數(shù)據(jù)”文章能幫助大家解決問(wèn)題。

背景

由于不同乙方對(duì)服務(wù)商業(yè)務(wù)接口字段理解不一致，導(dǎo)致線上上千萬(wàn)數(shù)據(jù)量數(shù)據(jù)存在問(wèn)題，為了修復(fù)數(shù)據(jù)，通過(guò) Python 腳本進(jìn)行修改

知識(shí)點(diǎn)

Python3、線程池、pymysql、CSV 文件操作、requests

拓展

當(dāng)我們程序在使用到線程、進(jìn)程或協(xié)程的時(shí)候，以下三個(gè)知識(shí)點(diǎn)可以先做個(gè)基本認(rèn)知

CPU 密集型、IO 密集型、GIL 全局解釋器鎖

庫(kù)

pip3 install requests

pip3 install pymysql

流程

實(shí)現(xiàn)代碼

# -*- coding:utf-8 -*-
# @FileName:grade_update.py
# @Desc    :在一臺(tái)超級(jí)計(jì)算機(jī)上運(yùn)行過(guò)的牛逼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():
    """
    數(shù)據(jù)庫(kù)連接
    """
    return pymysql.connect(host="localhost",
                         port=3306,
                         user="admin",
                         password="admin",
                         database="test"
                         )
def oprationdb(grade_id, member_id):
  """
  操作數(shù)據(jù)庫(kù)
  """
    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數(shù)據(jù)庫(kù)異常：", e)
    db.close()
    return True
def interface(rows, thead):
  """
  調(diào)用第三方接口
  """
    print(f"處理數(shù)據(jù)行數(shù)--->{thead}----數(shù)據(jù)--->{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"調(diào)用異常:{e}")
def read_csv():
    import csv
    # db = writ_mysql()
    #線程數(shù)
    MAX_WORKERS=5
    with ThreadPoolExecutor(MAX_WORKERS) as pool:
        with open(path + '/file/result2_colu.csv', 'r', newline='', encoding='utf-8') as f:
            #set() 函數(shù)創(chuàng)建無(wú)序不重復(fù)元素集
            seq_notdone = set()
            seq_done = set()
            # 使用csv的reader()方法，創(chuàng)建一個(gè)reader對(duì)象
            reader = csv.reader(f)
            n = 0
            for row in reader:
                n += 1
                # 遍歷reader對(duì)象的每一行
                try:
                    seq_notdone.add(pool.submit(interface, rows=row, thead=n))
                    if len(seq_notdone) >= MAX_WORKERS:
                        #FIRST_COMPLETED文檔說(shuō)明 -- 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"解析結(jié)果出錯(cuò)：{e}")
    # db.close()
    return "完成"
if __name__ == '__main__':
    read_csv()

解釋

引入線程池庫(kù)

from concurrent.futures import ThreadPoolExecutor,FIRST_COMPLETED,wait

pool.submit(interface, rows=row, thead=n)

提交任務(wù)，interface 調(diào)用的函數(shù)，rows、thead 為 interface() 函數(shù)的入?yún)?/p>

任務(wù)持續(xù)提交，線程池通過(guò) MAX_WORKERS 定義的線程數(shù)持續(xù)消費(fèi)

說(shuō)明像這種 I/O 密集型的操作腳本適合使用多線程，如果是 CPU 密集型建議使用進(jìn)行，根據(jù)機(jī)器核數(shù)進(jìn)行配置

關(guān)于“怎么使用python3線程池ThreadPoolExecutor處理csv文件數(shù)據(jù)”的內(nèi)容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識(shí)，可以關(guān)注億速云行業(yè)資訊頻道，小編每天都會(huì)為大家更新不同的知識(shí)點(diǎn)。