python - 如何让 concurrent.futures ProcessPoolExecutor 与字典一起工作？
问题描述
我在 youtube 上观看了 python 多处理教程，这是链接https://www.youtube.com/watch?v=fKl2JW_qrso&t=2316s&ab_channel=CoreySchafer
然后，我尝试在我的代码中应用该方法，这是我在应用多处理之前的代码：
import requests
from bs4 import BeautifulSoup
import pandas as pd
import investpy
from pandas import Timestamp
import json
from pandas.io.json import json_normalize
import time
def getCurrency_data():
user_agent = "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.88 Safari/537.37"
links = {"USD-IDR":"https://www.investing.com/currencies/usd-idr",
"USD-JPY":"https://www.investing.com/currencies/usd-jpy",
"USD-CNY":"https://www.investing.com/currencies/usd-cny",
"USD-EUR":"https://www.investing.com/currencies/usd-eur",
"USD-SGD":"https://www.investing.com/currencies/usd-sgd",
"USD-THB":"https://www.investing.com/currencies/usd-thb",
"USD-MXN":"https://www.investing.com/currencies/usd-mxn",
"USD-MYR":"https://www.investing.com/currencies/usd-myr",
"USD-KRW":"https://www.investing.com/currencies/usd-krw",
"USD-INR":"https://www.investing.com/currencies/usd-inr"}
df_currency= pd.DataFrame(columns =['Currency', 'Current', 'Change', 'Prev. Close', 'Open', '1 Year Change'])
currency=[]
current=[]
change=[]
prev_close=[]
open_=[]
oneyear_change=[]
for key, value in links.items():
data = requests.get(value, headers={'User-Agent': user_agent})
soup = BeautifulSoup(data.content, 'html.parser')
tags1 = soup.find_all('div', {'class':'top bold inlineblock'})
span_tag = []
for div in tags1:
spans = div.find_all('span')
for span in spans:
x = span.text
span_tag.append(x)
current_tmp = span_tag[0]
change_tmp = span_tag[1]
current.append(current_tmp)
change.append(change_tmp)
currency_tmp = key
currency.append(currency_tmp)
cur = []
tags2 = soup.find('div', {'class':'clear overviewDataTable overviewDataTableWithTooltip'})
for a in tags2.findAll('div', {'class':'first inlineblock'}):
for b in a.findAll('span', {'class':'float_lang_base_2 bold'}):
cur.append(b.text)
prevclose_tmp = cur[0]
open_tmp = cur[1]
oneyearchange_tmp = cur[2]
prev_close.append(prevclose_tmp)
open_.append(open_tmp)
oneyear_change.append(oneyearchange_tmp)
df_currency["Currency"] = currency
df_currency["Current"] = current
df_currency["Change"] = change
df_currency["Prev. Close"] = prev_close
df_currency["Open"] = open_
df_currency["1 Year Change"] = oneyear_change
return(df_currency)
好吧，但我很困惑。在那个视频中，输入是一个列表，而我使用字典......这是应用多处理后的代码：
import requests
from bs4 import BeautifulSoup
import pandas as pd
import investpy
from pandas import Timestamp
import json
from pandas.io.json import json_normalize
import time
import concurrent.futures
t1 = time.perf_counter()
links = {"USD-IDR":"https://www.investing.com/currencies/usd-idr",
"USD-JPY":"https://www.investing.com/currencies/usd-jpy",
"USD-CNY":"https://www.investing.com/currencies/usd-cny",
"USD-EUR":"https://www.investing.com/currencies/usd-eur",
"USD-SGD":"https://www.investing.com/currencies/usd-sgd",
"USD-THB":"https://www.investing.com/currencies/usd-thb",
"USD-MXN":"https://www.investing.com/currencies/usd-mxn",
"USD-MYR":"https://www.investing.com/currencies/usd-myr",
"USD-KRW":"https://www.investing.com/currencies/usd-krw",
"USD-INR":"https://www.investing.com/currencies/usd-inr"}
def getCurrency_data(link):
user_agent = "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.88 Safari/537.37"
df_currency= pd.DataFrame(columns =['Currency', 'Current', 'Change', 'Prev. Close', 'Open', '1 Year Change'])
currency=[]
current=[]
change=[]
prev_close=[]
open_=[]
oneyear_change=[]
for key, value in links.items():
data = requests.get(value, headers={'User-Agent': user_agent})
soup = BeautifulSoup(data.content, 'html.parser')
tags1 = soup.find_all('div', {'class':'top bold inlineblock'})
span_tag = []
for div in tags1:
spans = div.find_all('span')
for span in spans:
x = span.text
span_tag.append(x)
current_tmp = span_tag[0]
change_tmp = span_tag[1]
current.append(current_tmp)
change.append(change_tmp)
currency_tmp = key
currency.append(currency_tmp)
cur = []
tags2 = soup.find('div', {'class':'clear overviewDataTable overviewDataTableWithTooltip'})
for a in tags2.findAll('div', {'class':'first inlineblock'}):
for b in a.findAll('span', {'class':'float_lang_base_2 bold'}):
cur.append(b.text)
prevclose_tmp = cur[0]
open_tmp = cur[1]
oneyearchange_tmp = cur[2]
prev_close.append(prevclose_tmp)
open_.append(open_tmp)
oneyear_change.append(oneyearchange_tmp)
df_currency["Currency"] = currency
df_currency["Current"] = current
df_currency["Change"] = change
df_currency["Prev. Close"] = prev_close
df_currency["Open"] = open_
df_currency["1 Year Change"] = oneyear_change
return(df_currency)
with concurrent.futures.ProcessPoolExecutor() as executor:
executor.map(getCurrency_data, links)
t2 = time.perf_counter()
print(f'Finished in {t2-t1} secondes')
我的代码有问题吗？我已经尝试过了，但没有任何结果。谢谢你。
解决方案
这个问题也可以使用多线程而不是多处理来解决，因为大部分时间都花在getCurrency_data等待数据从您的requests.get请求中返回，因此竞争全局解释器锁的线程之间几乎没有争用。但是由于对返回的数据进行了一些 CPU 密集型处理，BeautifulSoup因此 GIL 总会存在一些争用，这表明：
(1) 多处理可能会比多线程执行得稍好，但前提是您创建的进程数量与您必须检索的 URL 数量一样多，以反映大多数时间您的“工作者”函数正在等待的事实，并且 (2) 您应该使用一个requests.Session实例来检索 URL，因为您的所有 URL 都针对同一个网站，这样做可以提高效率。
要将您的程序转换为多处理或多线程（尝试两种方式 - 您只需更改ProcessPoolExecutor为ThreadPoolExecutor，但我发现多处理的性能略高一些），函数getCurrency_data 应该只处理单个 URL 并将数据返回到主进程它已检索。然后，主进程应该累积所有子进程返回的数据并初始化数据帧：
import requests
from bs4 import BeautifulSoup
import pandas as pd
#import investpy
#from pandas import Timestamp
#import json
#from pandas.io.json import json_normalize
import time
import concurrent.futures
from functools import partial
def getCurrency_data(session, item):
key, value = item
data = session.get(value)
soup = BeautifulSoup(data.content, 'html.parser')
tags1 = soup.find_all('div', {'class':'top bold inlineblock'})
span_tag = []
for div in tags1:
spans = div.find_all('span')
for span in spans:
x = span.text
span_tag.append(x)
current_tmp = span_tag[0]
change_tmp = span_tag[1]
currency_tmp = key
cur = []
tags2 = soup.find('div', {'class':'clear overviewDataTable overviewDataTableWithTooltip'})
for a in tags2.findAll('div', {'class':'first inlineblock'}):
for b in a.findAll('span', {'class':'float_lang_base_2 bold'}):
cur.append(b.text)
prevclose_tmp = cur[0]
open_tmp = cur[1]
oneyearchange_tmp = cur[2]
return currency_tmp, current_tmp, change_tmp, prevclose_tmp, open_tmp, oneyearchange_tmp
def main():
t1 = time.perf_counter()
links = {"USD-IDR":"https://www.investing.com/currencies/usd-idr",
"USD-JPY":"https://www.investing.com/currencies/usd-jpy",
"USD-CNY":"https://www.investing.com/currencies/usd-cny",
"USD-EUR":"https://www.investing.com/currencies/usd-eur",
"USD-SGD":"https://www.investing.com/currencies/usd-sgd",
"USD-THB":"https://www.investing.com/currencies/usd-thb",
"USD-MXN":"https://www.investing.com/currencies/usd-mxn",
"USD-MYR":"https://www.investing.com/currencies/usd-myr",
"USD-KRW":"https://www.investing.com/currencies/usd-krw",
"USD-INR":"https://www.investing.com/currencies/usd-inr"}
with requests.Session() as session:
user_agent = "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.88 Safari/537.37"
session.headers = {'User-Agent': user_agent}
currency = []
current = []
change = []
prev_close = []
open_ = []
oneyear_change = []
with concurrent.futures.ProcessPoolExecutor(max_workers=len(links)) as executor:
for return_value in executor.map(partial(getCurrency_data, session), links.items()):
currency_tmp, current_tmp, change_tmp, prevclose_tmp, open_tmp, oneyearchange_tmp = return_value
currency.append(currency_tmp)
current.append(current_tmp)
change.append(change_tmp)
prev_close.append(prevclose_tmp)
open_.append(open_tmp)
oneyear_change.append(oneyearchange_tmp)
df_currency = pd.DataFrame(columns=['Currency', 'Current', 'Change', 'Prev. Close', 'Open', '1 Year Change'])
df_currency["Currency"] = currency
df_currency["Current"] = current
df_currency["Change"] = change
df_currency["Prev. Close"] = prev_close
df_currency["Open"] = open_
df_currency["1 Year Change"] = oneyear_change
t2 = time.perf_counter()
print(f'Finished in {t2-t1} seconds')
print(df_currency)
# Required for Windows:
if __name__ == '__main__':
main()
印刷：
Finished in 4.4468559 seconds
Currency Current Change Prev. Close Open 1 Year Change
0 USD-IDR 14,452.5 +52.5 14,400 14,450 - 2.49%
1 USD-JPY 109.81 +0.09 109.72 109.73 3.47%
2 USD-CNY 6.5006 +0.0064 6.4942 6.4951 - 6.13%
3 USD-EUR 0.8564 0.0001 0.8563 0.8565 1.33%
4 USD-SGD 1.3628 -0.0014 1.3643 1.364 - 0.44%
5 USD-THB 33.370 +0.020 33.35 33.34 6.54%
6 USD-MXN 20.3829 +0.2309 20.152 20.152 - 8.88%
7 USD-MYR 4.2375 +0.0005 4.237 4.2395 1.67%
8 USD-KRW 1,182.31 +6.03 1,176.28 1,175.46 - 0.69%
9 USD-INR 74.400 +0.030 74.37 74.38 - 0.62%
两全其美的
由于创建进程有相当多的开销，因此将需要完成的工作分为主要是 I/O（即检索 URL）和主要是 CPU（解析和处理检索到的 HTML 文档）是最有效的前者使用多线程，后者使用多处理。这样，您就不会创建比实际需要更多的流程。
和以前一样，多线程池的大小应该等于需要检索的 URL 的数量（只要这个数量不是很大；创建数百个线程应该不是问题），并且多处理池的大小应该最多使用您拥有的 CPU 内核数。因此，我们创建了两个池并将多处理池传递给我们的工作函数，该函数检索 URL，然后将数据提交到多处理池以解析和处理该数据。
在下面的代码中，我正在创建一个多处理池大小，该大小等于我拥有的物理核心数 (4)，这是我拥有的逻辑核心数的一半，如果我没有指定，这将是默认池大小一个尺寸。要动态确定此值，您可以psutil从PyPI存储库安装包：
import requests
from bs4 import BeautifulSoup
import pandas as pd
#import investpy
#from pandas import Timestamp
#import json
#from pandas.io.json import json_normalize
import time
import concurrent.futures
from functools import partial
import psutil
def process_data(key, data):
soup = BeautifulSoup(data, 'html.parser')
tags1 = soup.find_all('div', {'class':'top bold inlineblock'})
span_tag = []
for div in tags1:
spans = div.find_all('span')
for span in spans:
x = span.text
span_tag.append(x)
current_tmp = span_tag[0]
change_tmp = span_tag[1]
currency_tmp = key
cur = []
tags2 = soup.find('div', {'class':'clear overviewDataTable overviewDataTableWithTooltip'})
for a in tags2.findAll('div', {'class':'first inlineblock'}):
for b in a.findAll('span', {'class':'float_lang_base_2 bold'}):
cur.append(b.text)
prevclose_tmp = cur[0]
open_tmp = cur[1]
oneyearchange_tmp = cur[2]
return currency_tmp, current_tmp, change_tmp, prevclose_tmp, open_tmp, oneyearchange_tmp
def getCurrency_data(session, pool_executor, item):
key, value = item
data = session.get(value)
f = pool_executor.submit(process_data, key, data.content)
return f.result()
def main():
t1 = time.perf_counter()
links = {"USD-IDR":"https://www.investing.com/currencies/usd-idr",
"USD-JPY":"https://www.investing.com/currencies/usd-jpy",
"USD-CNY":"https://www.investing.com/currencies/usd-cny",
"USD-EUR":"https://www.investing.com/currencies/usd-eur",
"USD-SGD":"https://www.investing.com/currencies/usd-sgd",
"USD-THB":"https://www.investing.com/currencies/usd-thb",
"USD-MXN":"https://www.investing.com/currencies/usd-mxn",
"USD-MYR":"https://www.investing.com/currencies/usd-myr",
"USD-KRW":"https://www.investing.com/currencies/usd-krw",
"USD-INR":"https://www.investing.com/currencies/usd-inr"}
with requests.Session() as session:
user_agent = "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.88 Safari/537.37"
session.headers = {'User-Agent': user_agent}
currency = []
current = []
change = []
prev_close = []
open_ = []
oneyear_change = []
with concurrent.futures.ProcessPoolExecutor(psutil.cpu_count(logical=False)) as pool_executor, \
concurrent.futures.ThreadPoolExecutor(max_workers=len(links)) as executor:
for return_value in executor.map(partial(getCurrency_data, session, pool_executor), links.items()):
currency_tmp, current_tmp, change_tmp, prevclose_tmp, open_tmp, oneyearchange_tmp = return_value
currency.append(currency_tmp)
current.append(current_tmp)
change.append(change_tmp)
prev_close.append(prevclose_tmp)
open_.append(open_tmp)
oneyear_change.append(oneyearchange_tmp)
df_currency = pd.DataFrame(columns=['Currency', 'Current', 'Change', 'Prev. Close', 'Open', '1 Year Change'])
df_currency["Currency"] = currency
df_currency["Current"] = current
df_currency["Change"] = change
df_currency["Prev. Close"] = prev_close
df_currency["Open"] = open_
df_currency["1 Year Change"] = oneyear_change
t2 = time.perf_counter()
print(f'Finished in {t2-t1} seconds')
print(df_currency)
# Required for Windows:
if __name__ == '__main__':
main()
印刷：
Finished in 3.5800665 seconds
Currency Current Change Prev. Close Open 1 Year Change
0 USD-IDR 14,452.5 +52.5 14,400 14,450 - 2.49%
1 USD-JPY 109.81 +0.09 109.72 109.73 3.47%
2 USD-CNY 6.5015 +0.0073 6.4942 6.4951 - 6.13%
3 USD-EUR 0.8545 -0.0018 0.8563 0.8565 1.33%
4 USD-SGD 1.3615 -0.0027 1.3643 1.364 - 0.44%
5 USD-THB 33.360 +0.010 33.35 33.34 6.54%
6 USD-MXN 20.4000 +0.2480 20.152 20.152 - 8.88%
7 USD-MYR 4.2375 +0.0005 4.237 4.2395 1.67%
8 USD-KRW 1,177.58 +1.30 1,176.28 1,175.46 - 0.69%
9 USD-INR 74.352 -0.018 74.37 74.38 - 0.62%
推荐阅读
- python - 如何将数字添加到 s3bucket 和 postgrsql 中的文件名
- vb.net - 如何从第 3 方 exe 获取唯一的进程 ID
- python - 如何在python中找到点在曲面上的垂直投影
- r - 如何使用 R 和 rvest 轮换代理和 IP 地址
- flutter - 不能无条件调用运算符“*”，因为接收者可以为“null”。尝试向目标添加空检查（'！'）
- if-statement - SAS 保留语句组
- google-analytics - 向谷歌分析报告转化
- php - 面对419 | 即使表单中存在 CSRF，页面也已过期
- html - 具有 2 个重要的 CSS 优先级
- javascript - javascript websocket : 每个设备一个 websocket 或所有设备一个 websocket