并行异常检测实现报错：ValueError: No objects to concatenate

我编写了一个Python脚本，通过concurrent.futures对多组织的多种异常类型（amount、coa、payee_name）并发执行异常检测。脚本按功能拆分函数，每个函数负责检测单个组织内特定类型的异常。

报错发生在detect_anomalies_for_org函数中拼接异常DataFrames的代码行：

return pd.concat(anomalies_list)

具体报错：

ValueError: No objects to concatenate

并行处理代码 #

result_list = []
    
with concurrent.futures.ProcessPoolExecutor(max_workers=num_cpus) as executor:
    futures = []
    for Org_id in dataset['Org_id'].unique():
        future = executor.submit(detect_anomalies_for_org, dataset.copy(), Org_id, anomaly_type_sets)
        futures.append(future)
        print(f"Submitted task for Org_id: {Org_id}")
    
    concurrent.futures.wait(futures)
    
    for future in futures:
        result = future.result()
        if not result.empty:
            result_list.append(result)
   
# Concatenate non-empty DataFrames
if result_list:
    merged_anomalies = pd.concat(result_list)
else:
    merged_anomalies = pd.DataFrame(columns=['id', 'Txn_id', 'Org_id', 'Client_id', 'AnomalyType', 'AnomalyScore', 'AnomalyCategory'])
    
# Save the merged anomalies to a CSV file
merged_anomalies.to_csv(output_file_path, index=False)
    
elapsed_time = time.time() - start_time
print(f"Execution time: {elapsed_time:.2f} seconds")

报错完整输出 #

Traceback (most recent call last):
  File "/home/ec2-user/anaconda3/envs/python3/lib/python3.10/concurrent/futures/process.py", line 246, in _process_worker
    r = call_item.fn(*call_item.args, **call_item.kwargs)
  File "/tmp/ipykernel_10936/3957857038.py", line 177, in detect_anomalies_for_org
    return pd.concat(anomalies_list)
  File "/home/ec2-user/anaconda3/envs/python3/lib/python3.10/site-packages/pandas/core/reshape/concat.py", line 372, in concat
    op = _Concatenator(
  File "/home/ec2-user/anaconda3/envs/python3/lib/python3.10/site-packages/pandas/core/reshape/concat.py", line 429, in __init__
    raise ValueError("No objects to concatenate")
ValueError: No objects to concatenate
"""

The above exception was the direct cause of the following exception:

ValueError                                Traceback (most recent call last)
Cell In[3], line 202
    199 concurrent.futures.wait(futures)
    201 for future in futures:
--> 202     result = future.result()
    203     if not result.empty:
    204         result_list.append(result)

File ~/anaconda3/envs/python3/lib/python3.10/concurrent/futures/_base.py:451, in Future.result(self, timeout)
    449     raise CancelledError()
    450 elif self._state == FINISHED:
--> 451     return self.__get_result()
    453 self._condition.wait(timeout)
    455 if self._state in [CANCELLED, CANCELLED_AND_NOTIFIED]:

File ~/anaconda3/envs/python3/lib/python3.10/concurrent/futures/_base.py:403, in Future.__get_result(self)
    401 if self._exception:
    402     try:
--> 403         raise self._exception
    404     finally:
    405         # Break a reference cycle with the exception in self._exception
    406         self = None

ValueError: No objects to concatenate

已尝试的解决步骤 #

• 验证输入数据集（dataset）包含预期值；
• 检查基于Isolation Forest的异常检测逻辑及中间结果；
• 尝试处理无异常的边缘情况。

错误核心是detect_anomalies_for_org函数内的anomalies_list为空，直接调用pd.concat()触发异常。需要在函数内部先判断列表状态，再返回对应内容：

修改detect_anomalies_for_org的返回逻辑：

def detect_anomalies_for_org(dataset, org_id, anomaly_type_sets):
    anomalies_list = []
    # 原异常检测逻辑：逐个类型检测并将结果添加到anomalies_list...
    
    # 处理空列表场景
    if anomalies_list:
        return pd.concat(anomalies_list)
    else:
        # 返回与正常结果结构一致的空DataFrame
        return pd.DataFrame(columns=['id', 'Txn_id', 'Org_id', 'Client_id', 'AnomalyType', 'AnomalyScore', 'AnomalyCategory'])

补充说明 #

1. 外层判断无效的原因：外层是在获取future.result()后才判断结果是否为空，但函数内部抛出异常会直接终止子任务，根本走不到外层判断步骤，必须在子任务内部处理空列表。
2. 空DataFrame的列名必须与正常检测结果完全一致，避免后续拼接出现列不匹配问题。
3. 若某组织所有异常类型检测均无结果，函数返回空DataFrame，外层if not result.empty会自动跳过添加到result_list，不影响最终拼接逻辑。

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