Pandas中如何实现数据的递归过滤？复杂条件处理-Python教程-PHP中文网

Pandas中如何实现数据的递归过滤？复杂条件处理

看不見的法師

发布： 2025-07-29 13:37:01

原创

634人浏览过

pandas中实现数据的递归过滤，核心在于定义清晰的过滤条件和终止条件，并通过自定义函数和循环结构不断应用筛选规则。1. 首先，定义递归函数接收dataframe并返回过滤后的结果；2. 然后，使用循环不断调用该函数，直到满足终止条件或数据集不再变化；3. 对于复杂条件，可分解问题并结合自定义函数与逻辑运算符进行组合；4. 为避免递归深度过大，可采用迭代方法（如队列）替代递归；5. 处理循环依赖时，使用集合记录已访问节点以防止无限循环；6. 优化性能方面，可使用向量化操作、query方法、merge操作或并行处理技术提升效率。示例展示了如何查找所有向指定经理汇报的员工，以及如何扩展支持绩效评分过滤，并提供了使用迭代代替递归的实现方式。

Pandas中如何实现数据的递归过滤？复杂条件处理

Pandas中实现数据的递归过滤，本质上是在满足一系列复杂条件的情况下，重复应用筛选规则，直到数据集不再发生变化或满足特定终止条件。这通常涉及到自定义函数、循环结构，以及对Pandas DataFrame的灵活操作。

解决方案

递归过滤的核心在于定义清晰的过滤条件和终止条件。首先，你需要一个函数，该函数接收DataFrame作为输入，应用过滤条件，并返回过滤后的DataFrame。然后，你需要一个循环，在这个循环中不断调用这个函数，直到满足终止条件。

一个简单的例子是，假设你有一个DataFrame，其中包含员工信息，你想过滤出所有直接或间接向某个特定经理汇报的员工。这可以通过递归查找汇报关系来实现。

import pandas as pd

def recursive_filter(df, manager_id, report_id_col, manager_id_col, result=None):
    &quot;&quot;&quot;
    递归过滤员工，找到所有向指定经理汇报的员工（直接或间接）。

    Args:
        df: Pandas DataFrame，包含员工信息。
        manager_id: 要查找的经理ID。
        report_id_col: 员工ID列名。
        manager_id_col: 经理ID列名。
        result: 用于累积结果的DataFrame（初始为None）。

    Returns:
        包含所有向指定经理汇报的员工的DataFrame。
    &quot;&quot;&quot;
    if result is None:
        result = pd.DataFrame(columns=df.columns)

    direct_reports = df[df[manager_id_col] == manager_id]
    if direct_reports.empty:
        return result

    result = pd.concat([result, direct_reports], ignore_index=True)
    for employee_id in direct_reports[report_id_col]:
        result = recursive_filter(df, employee_id, report_id_col, manager_id_col, result)

    return result.drop_duplicates()


# 示例数据
data = {'employee_id': [1, 2, 3, 4, 5, 6, 7],
        'manager_id': [None, 1, 1, 2, 2, 3, 4],
        'name': ['A', 'B', 'C', 'D', 'E', 'F', 'G']}
df = pd.DataFrame(data)

# 查找所有向经理1汇报的员工
manager_id_to_find = 1
reports = recursive_filter(df, manager_id_to_find, 'employee_id', 'manager_id')
print(reports)

登录后复制

这个例子展示了如何使用递归函数来查找所有向指定经理汇报的员工。关键点在于：

定义递归函数： recursive_filter函数接收DataFrame、经理ID、员工ID列名和经理ID列名作为输入。
查找直接下属： 函数首先找到所有直接向指定经理汇报的员工。
递归调用： 然后，对于每个直接下属，递归调用recursive_filter函数，查找他们的下属。
累积结果： 使用result参数来累积结果，避免重复计算。
终止条件： 当没有直接下属时，递归结束。

如何处理更复杂的递归过滤条件？

更复杂的递归过滤条件可能涉及到多个列的组合、更复杂的逻辑判断，甚至需要访问外部数据源。在这种情况下，你需要：

分解问题： 将复杂的过滤条件分解为更小的、可管理的部分。
自定义函数： 为每个部分定义自定义函数，这些函数可以接收DataFrame的行作为输入，并返回True或False，表示该行是否满足条件。
组合条件： 使用Pandas的apply函数和逻辑运算符（例如&、|）将这些自定义函数组合起来，形成最终的过滤条件。
优化性能： 对于大型数据集，递归操作可能会非常慢。考虑使用其他技术，例如迭代方法、向量化操作或并行处理来优化性能。

例如，假设你想过滤出所有向经理1汇报，并且绩效评分高于某个阈值的员工。你可以定义一个自定义函数来检查绩效评分，并将它与汇报关系的过滤条件组合起来。

def check_performance(row, threshold):
    return row['performance_score'] > threshold

def recursive_filter_with_performance(df, manager_id, report_id_col, manager_id_col, performance_threshold, result=None):
    if result is None:
        result = pd.DataFrame(columns=df.columns)

    direct_reports = df[df[manager_id_col] == manager_id]
    if direct_reports.empty:
        return result

    # 增加绩效评分过滤
    direct_reports = direct_reports[direct_reports.apply(check_performance, axis=1, args=(performance_threshold,)) ]

    result = pd.concat([result, direct_reports], ignore_index=True)
    for employee_id in direct_reports[report_id_col]:
        result = recursive_filter_with_performance(df, employee_id, report_id_col, manager_id_col, performance_threshold, result)

    return result.drop_duplicates()

# 示例数据，增加performance_score列
data = {'employee_id': [1, 2, 3, 4, 5, 6, 7],
        'manager_id': [None, 1, 1, 2, 2, 3, 4],
        'name': ['A', 'B', 'C', 'D', 'E', 'F', 'G'],
        'performance_score': [80, 90, 75, 85, 95, 70, 80]}
df = pd.DataFrame(data)

# 查找所有向经理1汇报且绩效评分高于80的员工
manager_id_to_find = 1
performance_threshold = 80
reports = recursive_filter_with_performance(df, manager_id_to_find, 'employee_id', 'manager_id', performance_threshold)
print(reports)

登录后复制

如何避免递归深度过大导致的问题？

递归深度过大是递归操作的常见问题。在Pandas中，如果数据集非常大，递归深度可能会超过Python的默认限制，导致RecursionError。为了避免这个问题，你可以：

腾讯智影-AI数字人

基于AI数字人能力，实现7*24小时AI数字人直播带货，低成本实现直播业务快速增增，全天智能在线直播

查看详情

限制递归深度： 在递归函数中添加一个深度计数器，当深度超过某个阈值时，停止递归。
使用迭代方法： 将递归操作转换为迭代操作，避免使用递归。这通常需要使用循环结构和数据结构（例如队列或栈）来模拟递归过程。
增加递归深度限制： 虽然不推荐，但你可以使用sys.setrecursionlimit()函数来增加Python的递归深度限制。但是，这可能会导致其他问题，例如内存溢出。

例如，使用迭代方法来查找所有向指定经理汇报的员工：

def iterative_filter(df, manager_id, report_id_col, manager_id_col):
    &quot;&quot;&quot;
    使用迭代方法过滤员工，找到所有向指定经理汇报的员工（直接或间接）。

    Args:
        df: Pandas DataFrame，包含员工信息。
        manager_id: 要查找的经理ID。
        report_id_col: 员工ID列名。
        manager_id_col: 经理ID列名。

    Returns:
        包含所有向指定经理汇报的员工的DataFrame。
    &quot;&quot;&quot;
    result = pd.DataFrame(columns=df.columns)
    queue = [manager_id]
    visited = set()

    while queue:
        current_manager_id = queue.pop(0)
        if current_manager_id in visited:
            continue
        visited.add(current_manager_id)

        direct_reports = df[df[manager_id_col] == current_manager_id]
        if not direct_reports.empty:
            result = pd.concat([result, direct_reports], ignore_index=True)
            for employee_id in direct_reports[report_id_col]:
                queue.append(employee_id)

    return result.drop_duplicates()

# 示例数据
data = {'employee_id': [1, 2, 3, 4, 5, 6, 7],
        'manager_id': [None, 1, 1, 2, 2, 3, 4],
        'name': ['A', 'B', 'C', 'D', 'E', 'F', 'G']}
df = pd.DataFrame(data)

# 查找所有向经理1汇报的员工
manager_id_to_find = 1
reports = iterative_filter(df, manager_id_to_find, 'employee_id', 'manager_id')
print(reports)

登录后复制

这个例子展示了如何使用队列来模拟递归过程，避免了递归深度过大的问题。