Pandas多进程代码报错：'DataFrame'对象不可调用求助

嘿，我来帮你搞定这个问题！先直接揪出核心错误，再一步步修正你的代码。

1. 最直接的错误：你传了函数执行结果，而非函数本身 #

看你调用parallelize_dataframe的代码：

test = parallelize_dataframe(nearest_calc3, test_func(nearest_calc3, stores))

这里test_func(nearest_calc3, stores)会立刻执行并返回一个DataFrame，而parallelize_dataframe的第二个参数需要的是可调用的函数对象（比如test_func这个名字本身），不是已经运行完得到的结果。这就是为啥会报错TypeError: 'DataFrame' object is not callable——你让进程池去调用一个DataFrame，当然行不通啦！

2. 解决函数参数绑定问题 #

你的test_func需要两个参数：data和stores，但pool.map只会给传入的函数传递一个参数（即每个分片后的data）。所以我们得把stores提前绑定到test_func上，让它变成只接受一个参数的函数，这里用functools.partial就非常方便。

3. 修正apply的逻辑错误 #

原来的data['lat'].apply(...)是对lat列的每个元素单独调用nearest，但你传的args里的data['long']是整个列，这会导致每个lat值和所有long值计算距离，逻辑完全不对！你要的是对每行的lat和long调用nearest，所以得用data.apply(..., axis=1)来按行处理。

修正后的完整代码 #

import multiprocessing as mp
import numpy as np
import pandas as pd
from functools import partial  # 别忘了导入这个工具

nearest_calc3 = postcodes.head(1000).copy() # 测试用前1000行数据
partitions = 5
cores = mp.cpu_count()

def parallelize_dataframe(data, func):
    data_split = np.array_split(data, partitions)
    pool = mp.Pool(cores)
    data = pd.concat(pool.map(func, data_split))
    pool.close()
    pool.join()
    return data

def nearest(inlat1, inlon1, inlat2, inlon2, store, postcode):
    lat1 = np.radians(inlat1)
    lat2 = np.radians(inlat2)
    longdif = np.radians(inlon2 - inlon1)
    r = 6371.1009 # 距离单位为公里
    d = np.arccos(np.sin(lat1)*np.sin(lat2) + np.cos(lat1)*np.cos(lat2) * np.cos(longdif)) * r
    # 优化：直接用numpy找最小值，避免创建临时DataFrame
    min_idx = np.argmin(d)
    return f"{store.iloc[min_idx]}~{postcode.iloc[min_idx]}~{d[min_idx]}"

def test_func(data, stores):
    # 定义单行处理函数
    def process_row(row):
        return nearest(row['lat'], row['long'], stores['lat'], stores['long'], stores['index'], stores['pcds'])
    
    # 按行应用处理逻辑
    data['appended'] = data.apply(process_row, axis=1)
    # 拆分结果列
    data[['store','store_postcode','distance_km']] = data['appended'].str.split("~",expand=True)
    # 转换距离为数值类型，方便后续计算
    data['distance_km'] = pd.to_numeric(data['distance_km'])
    return data

if __name__ == '__main__':
    # 用partial把stores绑定到test_func，生成仅需data参数的新函数
    bound_func = partial(test_func, stores=stores)
    test = parallelize_dataframe(nearest_calc3, bound_func)

额外效率优化建议 #

• 我已经帮你优化了nearest函数，去掉了创建临时DataFrame的步骤，直接用numpy的argmin找最小值，运行速度会提升不少。
• 如果你的stores数据量很大，还可以试试scipy.spatial.KDTree做最近邻搜索，比逐行计算距离的效率高几个量级，尤其适合大数据场景。

内容的提问来源于stack exchange，提问作者Winby