并行化Pandas中基于重复产品名填充缺失值的函数

问题场景 #

手里有个1838379行的产品DataFrame，字段包括description、image_url、ean、product_name。因为存在重复的product_name，想利用同名称下的非空值填充对应字段的NaN，但最初写的逐行填充函数速度极慢，试了Pandarallel和Dask都踩了坑：

• Pandarallel报错，提示函数未定义且不支持lambda传参
• Dask进度条直接卡住，无响应

原实现代码 #

逐行填充函数 #

def fill_descriptions_images_ean_from_duplicates(row,train):
    import pandas as pd
    duplicated_rows = train.loc[train['product_name'] == row["product_name"]]
    if not duplicated_rows.empty:

        descriptions=duplicated_rows["description"].dropna()
        if not descriptions.empty:
            description=list(descriptions)[0]
            train.loc[train['product_name'] ==  row["product_name"], 'description',] = train.loc[train['product_name'] ==  row["product_name"], 'description'].fillna(description)

        images=duplicated_rows["image_url"].dropna()
        if not images.empty:
            
            image=list(images)[0]
            train.loc[train['product_name'] ==  row["product_name"], 'image_url',] = train.loc[train['product_name'] ==  row["product_name"], 'image_url'].fillna(image)

        eans=duplicated_rows["ean"].dropna()
        if not eans.empty:
            ean=list(eans)[0]
            train.loc[train['product_name'] ==  row["product_name"], 'ean',] = train.loc[train['product_name'] ==  row["product_name"], 'ean'].fillna(ean)

Pandarallel尝试代码 #

from pandarallel import pandarallel
import psutil

psutil.cpu_count(logical=False)

pandarallel.initialize()
train.parallel_apply(lambda row: fill_descriptions_images_ean_from_duplicates(row, train), axis=1)

Dask尝试代码 #

def process_partition(df_partition,train):
    df_partition.apply(lambda row: fill_descriptions_images_ean_from_duplicates(row, train), axis=1)
    return df_partition

import dask.dataframe as dd
from dask.diagnostics import ProgressBar
dask_train = dd.from_pandas(train, npartitions=7)
dask_df_applied = dask_train.map_partitions(lambda row: process_partition(row, train),meta=train.dtypes)
with ProgressBar():
    train=dask_df_applied.compute()

示例测试数据 #

import pandas as pd
import numpy as np

# Set the random seed for reproducibility
np.random.seed(42)

# Generate random data
data = {
    'product_name': ['Product A', 'Product B', 'Product B', 'Product C', 'Product D'] * 20,
    'description': np.random.choice([np.nan, 'Description'], size=100),
    'image_url': np.random.choice([np.nan, 'image_url'], size=100),
    'ean': np.random.choice([np.nan, 'EAN123456'], size=100)
}

# Create the DataFrame
train= pd.DataFrame(data)

最优解决方案：GroupBy+Transform矢量化填充 #

逐行apply本质是循环，百万级数据必然缓慢。直接用pandas分组矢量化操作，速度能提升几个数量级——核心思路是按product_name分组，取每组第一个非空值填充组内NaN。

实现代码 #

import numpy as np

def fill_missing_with_first_non_null(series):
    # 获取分组内第一个非空值，无有效值则保留NaN
    first_non_null = series.dropna().iloc[0] if not series.dropna().empty else np.nan
    # 填充组内所有缺失值
    return series.fillna(first_non_null)

# 批量处理需要填充的字段
fill_columns = ['description', 'image_url', 'ean']
train[fill_columns] = train.groupby('product_name')[fill_columns].transform(fill_missing_with_first_non_null)

代码说明 #

1. fill_missing_with_first_non_null：针对每个分组的字段序列，先提取第一个非空值，再用该值覆盖序列内所有NaN
2. groupby+transform：按产品名分组后对指定字段应用填充函数，返回与原DataFrame同维度的结果，直接覆盖原字段即可

该方案是pandas原生矢量化操作，无循环、无额外进程开销，百万级数据几秒就能完成，比并行处理效率更高。

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