在 Python 中，如何有效地将一个巨大的列表映射到对象？

Question

我正在解析称为 FCD 的巨大 XML 文件（> 400 MB，~7M 行），它是 SUMO 道路交通模拟器的输出。 我的目标是及时获取每辆车的位置。

示例 FCD 文件如下所示：

<fcd-export>
    <timestep time="0.00">
        <vehicle id="flow_0.0" x="605.79" y="1142.59"/>
        <vehicle id="flow_1.0" x="1911.72" y="2154.71"/>
        <vehicle id="flow_3.0" x="1907.24" y="2163.97"/>
    </timestep>
    <timestep time="0.10">
        <vehicle id="flow_0.0" x="605.81" y="1142.61"/>
        <vehicle id="flow_1.0" x="1911.70" y="2154.69"/>
        <vehicle id="flow_3.0" x="1907.22" y="2163.95"/>
    </timestep>
    <timestep time="0.20">
        <vehicle id="flow_0.0" x="605.85" y="1142.64"/>
        <vehicle id="flow_1.0" x="1911.66" y="2154.66"/>
        <vehicle id="flow_3.0" x="1907.18" y="2163.92"/>
    </timestep>
</fcd-export>

我正在将其解析为此类 dicts 的列表：{car_id, time, x, y} 使用 lxml 和 multiprocessing 库，它们工作正常，36000 个时间步长约 30 秒，XML 文件中约 7M 行。我在底部附加了parse_fcd() 函数。结果列表有 680 万个项目。

现在我需要映射那些[time, car_id, x, y] 项目以及时获得每辆车的所有位置。我创建了简单的类来存储这些数据：

class CarInfo:
    car_id: str
    time_locations: List[TimeLocation]

class TimeLocation:
    time: float
    x: float
    y: float

我尝试使用以下代码进行映射：

import multiprocessing as mp
from typing import List

def extract_car_infos_parallel(car_time_location_items: List[dict]) -> List[CarInfo]:
    car_ids = set(map(lambda item: item['car_id'], car_time_location_items))
    
    pool = mp.Pool(mp.cpu_count())
    car_infos = pool.starmap(extract_time_location_items_for_car, [(car_id, car_time_location_items) for car_id in car_ids])

    pool.close()
    pool.join()

    return car_infos
    
def extract_time_location_items_for_car(car_id: str, all_items: List[dict]) -> CarInfo: 
    car = CarInfo(car_id)
    items_for_car = list(filter(lambda item: item['car_id'] == car_id, all_items))
    car.time_locations = [TimeLocation(item['time'], item['x'], item['y']) for item in items_for_car]

    return car

代码运行大约 15 分钟并抛出 BrokenPipeError。我尝试将 dicts {car_id, time, x, y} 列表更改为具有这些值的列表列表，结果相同。

如何解决此问题以摆脱 BrokenPipeError 并加快速度？

PS：
这是从 XML 文件中解析 FCD 数据的代码：

from lxml import etree
from lxml.etree import XMLParser, parse
import multiprocessing as mp
from typing import List 

def parse_fcd_data_parallel(fcd_file: str) -> List[dict]:

    p = XMLParser(huge_tree=True)
    xml_data = parse(fcd_file, parser=p)
    fcd_data = xml_data.getroot()

    pool = mp.Pool(mp.cpu_count())

    results = pool.map(parse_fcd_timestep, [timestep for timestep in fcd_data])

    pool.close()
    pool.join()

    flatten_results = [item for sublist in results for item in sublist]
    return flatten_results

def parse_fcd_timestep(timestep) -> List[dict]:
    car_time_location_items: List[dict] = []

    time_stamp = timestep.get('time')

    for raw_car_info in timestep:
        car_id = raw_car_info.get('id')
        pos_x = raw_car_info.get('x')
        pos_y = raw_car_info.get('y')

        car_time_location_items.append({'car_id': car_id, 'time': time_stamp, 'x': pos_x, 'y': pos_y})

    return car_time_location_items

Answer 1

问题来自于低效的算法和数据结构。并行化操作没有多大帮助。

一方面，CPython GIL 会阻止您有效地为该代码使用多个线程。另一方面，由于许多需要序列化的 CPython 对象的进程间通信，多处理肯定不会加速代码（它实际上应该更慢且更不灵活）。

效率低下肯定来自list(filter(lambda item: item['car_id'] == car_id, all_items)) 行，它遍历每辆搜索到的汽车的所有项目。这会导致二次O(m n) 执行时间，其中m 是列表的大小（6 800 000 项），n 是汽车的数量（可能有数百辆）。一种更有效的解决方案是使用 Pandas 数据帧而不是 dict 列表（在内存和执行时间方面有巨大的开销）并执行groupby 操作它以准线性时间运行（因为它使用排序或基于散列的方法对汽车进行分组）。 Pandas 在内部使用 Numpy，它有效地将项目打包到内存中并使用本机整数/浮点数。这应该快几个数量级。