解决Python递归处理CSV时的最大递归深度超限问题

问题说明 #

处理包含NEW_ID、OLD_ID、CLOSE_DATE_TXT、NEW_DATE_TXT四列的CSV文件，需遵循以下规则：

• 若OLD_ID与NEW_ID相同，直接忽略该记录
• 若二者不同且OLD_ID存在于NEW_ID列中，需将后续关联的OLD_ID、NEW_DATE_TXT、CLOSE_DATE_TXT用--拼接，直到关联的OLD_ID不在NEW_ID列中为止

原递归Python脚本在数据量超过3000条时触发Runtime Error: maximum recursion depth exceeded，无法处理大规模数据，需替换为非递归或其他方案。

原递归代码问题 #

原递归函数通过深度递归遍历关联ID，但Python默认递归深度限制（默认约1000）导致数据量增大时栈溢出，无法处理长链ID的关联。

def find_benf(key,interim,close_date,start_date):
  if ids.get(key):
    val = ids[key]
    val_key = list(val.keys())
    val_key = val_key[0]
    close_date.append(val[str(val_key)]['CLOSE_DATE_TXT'])
    start_date.append(val[str(val_key)]['NEW_DATE_TXT'])
    interim.append(val_key)
    return find_benf(val_key,interim,close_date,start_date)
  else:
    return interim,start_date,close_date

id_list,start_date,close_date_list = find_benf(row[0],new_id,close_date,start_date)

解决方案 #

1. Python非递归实现 #

用循环替代递归，通过字典映射存储关联关系，遍历每个ID的关联链，避免栈溢出：

import pandas as pd

# 读取CSV数据
df = pd.read_csv('your_file.csv')

# 过滤掉OLD_ID与NEW_ID相同的记录
df = df[df['OLD_ID'] != df['NEW_ID']].reset_index(drop=True)

# 构建NEW_ID到对应行数据的映射字典，方便快速查找
id_map = df.set_index('NEW_ID').to_dict('index')

# 初始化结果列表
result = []

# 遍历每一行数据
for _, row in df.iterrows():
    current_id = row['NEW_ID']
    current_old = row['OLD_ID']
    current_close = row['CLOSE_DATE_TXT']
    current_new = row['NEW_DATE_TXT']
    
    # 存储后续关联的字段
    chain_old = [current_old]
    chain_close = [current_close]
    chain_new = [current_new]
    
    # 循环查找关联链，直到没有后续NEW_ID匹配
    next_id = current_old
    while next_id in id_map:
        next_row = id_map[next_id]
        chain_old.append(next_row['OLD_ID'])
        chain_close.append(next_row['CLOSE_DATE_TXT'])
        chain_new.append(next_row['NEW_DATE_TXT'])
        next_id = next_row['OLD_ID']
    
    # 拼接字段并添加到结果
    result.append({
        'NEW_ID': current_id,
        'OLD_ID': '--'.join(map(str, chain_old)),
        'CLOSE_DATE_TXT': '--'.join(chain_close),
        'NEW_DATE_TXT': '--'.join(chain_new)
    })

# 转换为DataFrame并保存
result_df = pd.DataFrame(result)
result_df.to_csv('output.csv', index=False)

2. SQL递归CTE实现 #

支持递归CTE的数据库（如MySQL 8.0+、PostgreSQL、SQL Server）可直接用递归查询实现，无需担心递归深度问题：

WITH RECURSIVE id_chain AS (
    -- 基础查询：初始行，过滤ID相同的记录
    SELECT
        NEW_ID,
        CAST(OLD_ID AS CHAR) AS OLD_ID_CHAIN,
        CAST(CLOSE_DATE_TXT AS CHAR) AS CLOSE_DATE_CHAIN,
        CAST(NEW_DATE_TXT AS CHAR) AS NEW_DATE_CHAIN,
        OLD_ID AS NEXT_ID
    FROM your_table
    WHERE OLD_ID != NEW_ID
    
    UNION ALL
    
    -- 递归查询：关联后续行
    SELECT
        ic.NEW_ID,
        CONCAT(ic.OLD_ID_CHAIN, '--', t.OLD_ID) AS OLD_ID_CHAIN,
        CONCAT(ic.CLOSE_DATE_CHAIN, '--', t.CLOSE_DATE_TXT) AS CLOSE_DATE_CHAIN,
        CONCAT(ic.NEW_DATE_CHAIN, '--', t.NEW_DATE_TXT) AS NEW_DATE_CHAIN,
        t.OLD_ID AS NEXT_ID
    FROM id_chain ic
    JOIN your_table t ON ic.NEXT_ID = t.NEW_ID
    WHERE t.OLD_ID != t.NEW_ID
),
-- 筛选每个NEW_ID的最长关联链（即最终聚合结果）
final_chain AS (
    SELECT
        NEW_ID,
        OLD_ID_CHAIN AS OLD_ID,
        CLOSE_DATE_CHAIN AS CLOSE_DATE_TXT,
        NEW_DATE_CHAIN AS NEW_DATE_TXT,
        ROW_NUMBER() OVER (PARTITION BY NEW_ID ORDER BY LENGTH(OLD_ID_CHAIN) DESC) AS rn
    FROM id_chain
)
SELECT NEW_ID, OLD_ID, CLOSE_DATE_TXT, NEW_DATE_TXT
FROM final_chain
WHERE rn = 1
ORDER BY NEW_ID;

3. Informatica ETL解决方案 #

使用Informatica的Hierarchy Builder转换或循环Lookup实现：

• 步骤1：源数据读取后，先过滤OLD_ID = NEW_ID的记录
• 步骤2：使用Lookup转换，以OLD_ID作为查找键，匹配NEW_ID列，获取后续关联的行数据
• 步骤3：通过Loop组件或Hierarchy Builder的层级遍历功能，迭代拼接OLD_ID、CLOSE_DATE_TXT、NEW_DATE_TXT字段，直到无匹配的NEW_ID
• 步骤4：将最终拼接后的字段写入目标表或文件

样例数据 #

输入数据 #

输出数据 #

内容的提问来源于stack exchange，提问作者Mr.Sappeswaran