python3線程池ThreadPoolExecutor處理csv文件數(shù)據(jù)
背景
由于不同乙方對(duì)服務(wù)商業(yè)務(wù)接口字段理解不一致,導(dǎo)致線上上千萬數(shù)據(jù)量數(shù)據(jù)存在問題,為了修復(fù)數(shù)據(jù),通過 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 全局解釋器鎖
庫
pip3 install requests
pip3 install pymysql
流程
實(shí)現(xiàn)代碼
# -*- coding:utf-8 -*- # @FileName:grade_update.py # @Desc :在一臺(tái)超級(jí)計(jì)算機(jī)上運(yùn)行過的牛逼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ù)庫連接 """ return pymysql.connect(host="localhost", port=3306, user="admin", password="admin", database="test" ) def oprationdb(grade_id, member_id): """ 操作數(shù)據(jù)庫 """ 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ù)庫異常:", 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)建無序不重復(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文檔說明 -- 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()
解釋
引入線程池庫
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ù)提交,線程池通過 MAX_WORKERS 定義的線程數(shù)持續(xù)消費(fèi)
說明像這種 I/O 密集型的操作腳本適合使用多線程,如果是 CPU 密集型建議使用進(jìn)行,根據(jù)機(jī)器核數(shù)進(jìn)行配置
以上就是python3線程池ThreadPoolExecutor處理csv文件數(shù)據(jù)的詳細(xì)內(nèi)容,更多關(guān)于python3 ThreadPoolExecutor處理csv的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
Pandas DataFrame 取一行數(shù)據(jù)會(huì)得到Series的方法
今天小編就為大家分享一篇Pandas DataFrame 取一行數(shù)據(jù)會(huì)得到Series的方法,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過來看看吧2018-11-11計(jì)算機(jī)二級(jí)python學(xué)習(xí)教程(1) 教大家如何學(xué)習(xí)python
這篇文章主要為大家詳細(xì)介紹了計(jì)算機(jī)二級(jí)python學(xué)習(xí)教程,具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2019-05-05Django 實(shí)現(xiàn)jwt認(rèn)證的示例
這篇文章主要介紹了Django 實(shí)現(xiàn)jwt 認(rèn)證的示例,幫助大家更好的理解和學(xué)習(xí)使用django,感興趣的朋友可以了解下2021-04-04Python用5行代碼實(shí)現(xiàn)批量摳圖的示例代碼
這篇文章主要介紹了Python用5行代碼實(shí)現(xiàn)批量摳圖的示例代碼,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-04-04使用Streamlit和Pandas實(shí)現(xiàn)帶有可點(diǎn)擊鏈接的數(shù)據(jù)表格
這篇文章主要為大家詳細(xì)介紹了如何利用?Streamlit?和?Pandas?在?Python?中創(chuàng)建一個(gè)帶有可點(diǎn)擊鏈接的數(shù)據(jù)表格,感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下2023-11-11python3使用sqlite3構(gòu)建本地持久化緩存的過程
日常python開發(fā)中會(huì)遇到數(shù)據(jù)持久化的問題,今天記錄下如何使用sqlite3進(jìn)行數(shù)據(jù)持久化,并提供示例代碼及數(shù)據(jù)查看工具,需要的朋友可以參考下2023-11-11