Python双循环列表推导式筛选结果重复，如何实现去重？

现有三组数据task_resources、req_tasks、rel_tasks，需从task_resources中筛选出同时满足以下条件的条目：

• tasks_id匹配rel_tasks与req_tasks关联后的对应ID
• qos_level匹配req_tasks中的对应级别

当前使用双循环列表推导式实现时，返回结果存在重复条目，需得到无重复的目标列表。

数据与当前实现代码 #

task_resources = [
  {
    "resource_id": "52a58a34-6b1b-49b7-b53d-2bc4bf72c172",
    "domain_name": "Nep",
    "tasks_id": "702148b2-4f21-4034-af8d-d11737f9f811",
    "qos_level": "default",
    "cpu_cores": 1,
    "memory_mb": 128,
    "storage_mb": 0,
    "cpu_flops": 19177,
    "gpu_flops": 0,
    "gpu_memory_mb": 0,
    "egress_network_bandwidth_gbps": 0,
    "ingress_network_bandwidth_gbps": 0,
    "create_time": "2024-11-13T12:13:52.192657"
  },
  {
    "resource_id": "8208915d-90b4-4a6c-a98c-e9eb965b4e6b",
    "domain_name": "Nep",
    "tasks_id": "702148b2-4f21-4034-af8d-d11737f9f811",
    "qos_level": "low",
    "cpu_cores": 1,
    "memory_mb": 128,
    "storage_mb": 0,
    "cpu_flops": 19177,
    "gpu_flops": 0,
    "gpu_memory_mb": 0,
    "egress_network_bandwidth_gbps": 0,
    "ingress_network_bandwidth_gbps": 0,
    "create_time": "2024-11-13T12:13:52.202177"
  },
  {
    "resource_id": "c34a4411-20d3-42eb-9f74-1336ab6ab024",
    "domain_name": "Nep",
    "tasks_id": "95bc75a6-b292-484d-a65d-86fae2cb6b6f",
    "qos_level": "default",
    "cpu_cores": 1,
    "memory_mb": 128,
    "storage_mb": 0,
    "cpu_flops": 19177,
    "gpu_flops": 0,
    "gpu_memory_mb": 0,
    "egress_network_bandwidth_gbps": 0,
    "ingress_network_bandwidth_gbps": 0,
    "create_time": "2024-11-13T12:13:52.197433"
  }
]

req_tasks = [
                {'task_name': 'iperf-server', 'qos_level': 'default'},
                {'task_name': 'iperf-client', 'qos_level': 'default'}
            ]

rel_tasks = [
    {
        'tasks_id': '702148b2-4f21-4034-af8d-d11737f9f811', 
        'domain_name': 'Nep', 
        'package_name': 'iperf-client', 
        'composite_key': 'Nep-iperf-client', 
        'version': 'v1', 
        'package_type': 'GENERIC', 
        'interoperability_tags': 'null', 
        'blacklisted_tags': 'null', 
        'task_create_time': '2024-11-13T12:13:52.190297'
    }, 
    {
        'tasks_id': '95bc75a6-b292-484d-a65d-86fae2cb6b6f', 
        'domain_name': 'Nep', 
        'package_name': 'iperf-server', 
        'composite_key': 'Nep-iperf-server', 
        'version': 'v1', 
        'package_type': 'GENERIC', 
        'interoperability_tags': 'null', 
        'blacklisted_tags': 'null', 
        'task_create_time': '2024-11-13T12:13:52.195040'
    }]

# 当前实现代码
task_ids = [
    d["tasks_id"]
    for d in rel_tasks
    for t in req_tasks
    if d["package_name"] == t["task_name"]
]

resources = [
    d
    for t in req_tasks
    for d in task_resources
    if d["tasks_id"] in task_ids and d["qos_level"] == t["qos_level"]
]
print(resources)

重复原因分析 #

当前resources列表推导式外层循环遍历req_tasks，每个req_task都会完整遍历一次task_resources并匹配条件。如果多个req_task的qos_level相同，同一个符合条件的task_resource会被多次选中，导致重复。比如示例中两个req_task的qos_level都是default，符合条件的资源会被重复添加两次。

解决方案 #

方案一：优化筛选逻辑，从根源避免重复 #

先构建tasks_id与对应qos_level的映射关系，再直接遍历task_resources匹配条件，每个资源仅被检查一次。

# 构建任务名称到qos_level的映射
req_qos_map = {t['task_name']: t['qos_level'] for t in req_tasks}

# 构建符合要求的tasks_id到对应qos_level的映射
task_id_qos_map = {
    rt['tasks_id']: req_qos_map[rt['package_name']]
    for rt in rel_tasks
    if rt['package_name'] in req_qos_map
}

# 筛选符合条件的资源，无重复
resources = [
    res for res in task_resources
    if res['tasks_id'] in task_id_qos_map and res['qos_level'] == task_id_qos_map[res['tasks_id']]
]

print(resources)

方案二：对现有结果去重 #

如果不想修改筛选逻辑，可基于资源的唯一标识（如resource_id）对结果去重：

task_ids = [
    d["tasks_id"]
    for d in rel_tasks
    for t in req_tasks
    if d["package_name"] == t["task_name"]
]

# 按原逻辑生成带重复的列表
resources_with_duplicates = [
    d
    for t in req_tasks
    for d in task_resources
    if d["tasks_id"] in task_ids and d["qos_level"] == t["qos_level"]
]

# 去重：用resource_id作为唯一标识
seen_ids = set()
resources = []
for res in resources_with_duplicates:
    res_id = res['resource_id']
    if res_id not in seen_ids:
        seen_ids.add(res_id)
        resources.append(res)

print(resources)

方案对比 #

• 方案一：从根源避免重复生成，执行效率更高，适合数据量较大的场景
• 方案二：修改成本低，直接基于现有代码调整，但数据量大时会产生额外内存开销

内容的提问来源于stack exchange，提问作者Mark R