为何Pandas的iterrows循环用列表推导和多线程无法提速？

我需要遍历Pandas DataFrame的每行，执行两项操作生成新的字符串字段。这些DataFrame来自多个大型文件，执行速度至关重要。

复现用的小型GeoDataFrame #

import json
import geopandas as gpd
import pandas as pd
from shapely.geometry import Polygon
from geolib import geohash as pgh
import time

coords1 = [[1.3093500680264314, 44.14597523454133],
[1.309362565037715, 44.14597542723667],
[1.3093750620490838, 44.1459756199306],
[1.3093875590605368, 44.145975812623114],
[1.3094000560720744, 44.14597600531418]]

coords2 = [[1.3123256948057254, 44.14597606298388],
[1.312338191827267, 44.145976255340116],
[1.3123506888488932, 44.14597644769494],
[1.312363185870604, 44.14597664004834],
[1.3123756828923991, 44.14597683240032]]

elev1 = [177.2, 177.1, 177.0, 177.0, 176.9]

elev2 = [164.3, 164.5, 164.7, 165.0, 165.3]

points1 = gpd.GeoDataFrame({'center': coords1, 'elevation': elev1})
points2 = gpd.GeoDataFrame({'center': coords2, 'elevation': elev2})

inputs = [points1, points2]

原始的iterrows实现 #

def asc_to_geohash_polygons(gdf, processed_geohashes):
    
    start_time1 = time.time()
    
    elevation_data=[]
    for index, row in gdf.iterrows():
        geohash_code = pgh.encode(row['center'][1], row['center'][0], precision=9)

        if geohash_code not in processed_geohashes:
            processed_geohashes.add(geohash_code)
            bounds = pgh.bounds(geohash_code)
            sw = bounds.sw
            ne = bounds.ne
            poly = Polygon([(sw.lon, sw.lat),
                            (ne.lon, sw.lat),
                            (ne.lon, ne.lat),
                            (sw.lon, ne.lat),
                            (sw.lon, sw.lat)])
            feature = {
                "type": "Feature",
                "properties": {
                    "geohash": geohash_code,
                    "elevation": row['elevation']
                },
                "geometry": poly.__geo_interface__
            }
            features.append(feature)
            elevation_data.append([geohash_code, row['elevation']])
    
    end_time1 = time.time()
    print(f"Timer 1: {end_time1 - start_time1} seconds")
            
    return features, elevation_data

调用函数的主逻辑 #

def process_all_files_in_folder(gdfs, output_geojson_path):
    all_features = []
    elevation_dict = {}  # 使用字典去重
    processed_geohashes = set()  # 全局集合跟踪已处理的geohash

    all_files = [f for f in gdfs]

    for file in all_files:
        temp_data, temp_elevations = asc_to_geohash_polygons(file, processed_geohashes)
        all_features.extend(temp_data)

        for geohash, elevation in temp_elevations:
            elevation_dict[geohash] = elevation


    all_elevations = [[geohash, elevation] for geohash, elevation in elevation_dict.items()]
    print(all_elevations)
    with open(output_geojson_path, 'w') as f:
        json.dump({"type": "FeatureCollection", "features": all_features}, f, indent=4)
        

process_all_files_in_folder(inputs, 'output.geojson')

计时后发现iterrows循环是耗时瓶颈，我尝试了三种优化方案，但效果都不如原始方法：

尝试的优化方案 #

(1) 使用.apply() #

def encode_geohash(lat_lon_pair):
    return pgh.encode(lat_lon_pair[1], lat_lon_pair[0], precision=9)
        
def get_feature(geohash, elevation):
    bounds = pgh.bounds(geohash)
    sw = bounds.sw
    ne = bounds.ne
    poly = Polygon([(sw.lon, sw.lat),
                    (ne.lon, sw.lat),
                    (ne.lon, ne.lat),
                    (sw.lon, ne.lat),
                    (sw.lon, sw.lat)])
    feature = {
        "type": "Feature",
        "properties": {
            "geohash": geohash,
            "elevation": elevation
        },
        "geometry": poly.__geo_interface__
    }
    return feature

geohash_data = gdf['center'].apply(encode_geohash)
elevation_dataframe = pd.concat([geohash_data.rename('geohash'), gdf['elevation']], axis=1)
elevation_data = elevation_dataframe.values.tolist()
features = elevation_dataframe.apply(lambda x: get_feature(x['geohash'], x['elevation']), axis=1).tolist()
processed_geohashes.update(geohash_data.tolist())

(2) 列表推导 #

geohash_data = gdf['center'].apply(encode_geohash)
        
gdf['geohash'] = geohash_data
        
features = [get_feature(geohash, elevation) for elevation, geohash in zip(gdf['elevation'], gdf['geohash'])]
        
elevation_data = [[geohash, elevation] for geohash, elevation in zip(gdf['geohash'], gdf['elevation'])]
        
processed_geohashes.update(geohash_data.tolist())

(3) 多线程 #

from concurrent.futures import ThreadPoolExecutor
from threading import Lock

features = []
elevation_data = []
hash_lock = Lock()

pool = ThreadPoolExecutor(10)

def add_to_list(row):
    geohash_code = pgh.encode(row['center'][1], row['center'][0], precision=9)
    with hash_lock:
        if geohash_code in processed_geohashes:
            return False
        processed_geohashes.add(geohash_code)
    bounds = pgh.bounds(geohash_code)
    sw = bounds.sw
    ne = bounds.ne
    poly = Polygon([(sw.lon, sw.lat),
                    (ne.lon, sw.lat),
                    (ne.lon, ne.lat),
                    (sw.lon, ne.lat),
                    (sw.lon, sw.lat)])
    feature = {
        "type": "Feature",
        "properties": {
            "geohash": geohash_code,
            "elevation": row['elevation']
        },
        "geometry": poly.__geo_interface__
    }
    features.append(feature)
    elevation_data.append([geohash_code, row['elevation']])
    return True

tasks = []
for index, row in gdf.iterrows():
    tasks.append(pool.submit(add_to_list, row))

for task in tasks:
    task.result()

我忽略了什么？我意识到向量化也是一种可能，但由于要生成字符串，可能帮助不大。

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