本文旨在解决在使用 Map 函数和 Partial 方法结合 ThreadPoolExecutor 时，for 循环只迭代一次的问题。通过分析问题代码，指出使用 enumerate 函数直接迭代 DataFrame 列的错误，并提供使用 itertuples 方法正确迭代 DataFrame 行的解决方案，确保循环能够处理所有数据。
在使用 ThreadPoolExecutor 的 map 函数结合 partial 方法处理 DataFrame 数据时，可能会遇到 for 循环只执行一次的意外行为。 这通常是由于对 DataFrame 的迭代方式不正确导致的。下面将详细分析问题原因并提供解决方案。
问题分析
原始代码中使用 enumerate 函数直接迭代 DataFrame 的某一列，例如 _df[[f'{_link_column}']]。 这种方式实际上迭代的是一个 DataFrame 对象，而不是预期的链接序列。 因此，循环只会执行一次，导致只有第一行数据被处理。
解决方案：使用 itertuples 方法
itertuples 方法是 pandas DataFrame 提供的一种高效的行迭代方式。 它将 DataFrame 的每一行转换为一个 namedtuple 对象，从而可以方便地访问每一列的值。
以下是修改后的 get_the_text 函数：
import pandas as pd
def get_the_text(_df, _firms: list, _link_column: str):
"""
发送请求以接收文章的文本
参数
----------
_df : DataFrame
返回
-------
dataframe with the text of the articles
"""
_df.reset_index(inplace=True)
print(_df)
for row in _df.itertuples(index=False):
link = getattr(row, f'{_link_column}')
print(link)
if link:
website_text = list()
try:
# 假设 send_two_requests 函数已定义
page_status_code, page_content, page_url = send_two_requests(link)
# 在这里添加处理 page_content 的代码
website_text.append(page_content) # 示例
except Exception as e:
print(f"Error processing link {link}: {e}")
# 在这里添加将 website_text 添加到 _df 的代码
return _df # 返回修改后的 DataFrame代码解释：
- _df.itertuples(index=False): 使用 itertuples 方法迭代 DataFrame 的每一行。 index=False 参数表示不包含索引列。
- row = _df.itertuples(index=False): 循环变量 row 现在是一个 namedtuple 对象，包含了 DataFrame 的每一列的值。
- link = getattr(row, f'{_link_column}'): 使用 getattr 函数获取 row 对象中名为 _link_column 的属性值，即链接地址。
- 错误处理: 增加 try...except 块来捕获和处理请求过程中可能发生的异常，避免程序崩溃。
- 返回 DataFrame: 确保函数返回修改后的 DataFrame。
完整示例代码:
import pandas as pd
from functools import partial
from concurrent.futures import ThreadPoolExecutor
import requests
from bs4 import BeautifulSoup
# 模拟 send_two_requests 函数
def send_two_requests(url):
try:
response = requests.get(url)
response.raise_for_status() # 检查请求是否成功
soup = BeautifulSoup(response.content, 'html.parser')
return response.status_code, soup.get_text(), url
except requests.exceptions.RequestException as e:
print(f"Request failed for {url}: {e}")
return None, None, url
def get_the_text(_df, _firms: list, _link_column: str):
"""
发送请求以接收文章的文本
参数
----------
_df : DataFrame
返回
-------
dataframe with the text of the articles
"""
_df.reset_index(inplace=True)
print(_df)
for row in _df.itertuples(index=False):
link = getattr(row, f'{_link_column}')
print(link)
if link:
website_text = list()
try:
page_status_code, page_content, page_url = send_two_requests(link)
# 在这里添加处理 page_content 的代码
if page_content:
website_text.append(page_content) # 示例
except Exception as e:
print(f"Error processing link {link}: {e}")
# 在这里添加将 website_text 添加到 _df 的代码，例如：
# _df.loc[_df[_link_column] == link, 'text'] = ' '.join(website_text) # 示例
return _df # 返回修改后的 DataFrame
# 示例数据
data = {
'index': [1366, 4767, 6140, 11898],
'DATE': ['2014-01-12', '2014-01-12', '2014-01-12', '2014-01-12'],
'SOURCES': ['go.com', 'bloomberg.com', 'latimes.com', 'usatoday.com'],
'SOURCEURLS': [
'http://abcnews.go.com/Business/wireStory/mercedes-recalls-372k-suvs-21445846',
'http://www.bloomberg.com/news/2014-01-12/vw-patent-application-shows-in-car-gas-heater.html',
'http://www.latimes.com/business/autos/la-fi-hy-autos-recall-mercedes-20140112-story.html',
'http://www.usatoday.com/story/money/cars/2014/01/12/mercedes-recall/4437279/'
],
'Tone': [-0.375235, -1.842752, 1.551724, 2.521008],
'Positive_Score': [2.626642, 1.228501, 3.275862, 3.361345],
'Negative_Score': [3.001876, 3.071253, 1.724138, 0.840336],
'Polarity': [5.628518, 4.299754, 5.0, 4.201681],
'Activity_Reference_Density': [22.326454, 18.918919, 22.931034, 19.327731],
'Self_Group_Reference_Density': [0.0, 0.0, 0.344828, 0.840336],
'Year': [2014, 2014, 2014, 2014],
'Month': [1, 1, 1, 1],
'Day': [12, 12, 12, 12],
'Hour': [0, 0, 0, 0],
'Minute': [0, 0, 0, 0],
'Second': [0, 0, 0, 0],
'Mentioned_firms': ['mercedes', 'vw', 'mercedes', 'mercedes'],
'text': ['', '', '', '']
}
# 创建 DataFrame
df = pd.DataFrame(data)
# 使用 ThreadPoolExecutor 和 partial
_link_column = 'SOURCEURLS'
_firms = ['mercedes', 'vw']
get_the_text_par = partial(get_the_text, _link_column=_link_column, _firms=_firms)
with ThreadPoolExecutor() as executor:
chunk_size = len(df) if len(df) < 10 else len(df) // 10
chunks = [df.iloc[i:i + chunk_size] for i in range(0, len(df), chunk_size)]
result = list(executor.map(get_the_text_par, chunks))
print("处理完成!")注意事项：
- 确保 send_two_requests 函数能够正确处理各种网络请求情况，并进行适当的错误处理。
- 根据实际需求，调整 chunk_size 的大小，以优化并行处理的效率。
- 根据实际需求，在 get_the_text 函数中添加将提取到的文本信息添加到 DataFrame 中的代码。
- 在实际应用中，可能需要添加更完善的错误处理机制，例如重试机制等。
总结
通过使用 itertuples 方法，可以正确地迭代 DataFrame 的每一行，从而解决在使用 Map 函数和 Partial 方法后 for 循环只执行一次的问题。 同时，需要注意错误处理和数据处理的细节，以确保代码的健壮性和可靠性。
