我正在尝试计算ID中第一个Sent日期/时间和最后一个Received日期/时间之间的差异，并按源和目标对它们进行分组。示例（名为test_subset）如下所示（但它有'000s行）：

| ID   | From | To   | leg_sent | leg_received | Source | Destination |
|:-----|:-----|:-----|:---------|:-------------|:-------|:------------|
| 1btr | ABC  | XYZ  | 08:22:23 | 08:22:41     | GB     | FR          |
| 1btr | XYZ  | DEF  | 08:22:49 | 08:23:05     | GB     | FR          |
| 2vyu | LMN  | JFK  | 14:35:11 | 14:35:23     | US     | DE          |
| 2vyu | JFK  | HIJ  | 14:35:35 | 14:35:48     | US     | DE          |
| 2vyu | HIJ  | TPQ  | 14:35:51 | 14:36:25     | US     | DE          |

在数据ID中，From和To行是传输的每个支路的传输ID，在某些情况下，它们可能只有2个支路，在其它情况下，可能有多达10个支路;需要计算的是第一次发送和最后一次接收之间的差值，这两个值不会在同一行上。Source和Destination在与同一ID相关的所有行中保持不变。
我现在有一段代码（如下所示），可以计算同一行上leg_sent和leg_received之间的差值，并按照From和To代码对差值进行分组：

pairs = (
test_subset
.withColumn('"leg_time"', F.datediff('seconds', F.col('"leg_sent"'), F.col('"leg_received"')))
.groupBy(['"From"', '"To"'])
.agg(
    F.count('*').alias('"num_paths"'),
    F.avg('"leg_time"').alias('"average_leg_time"'),
    F.min('"leg_time"').alias('"min_leg_time"'),
    F.max('"leg_time"').alias('"max_leg_time"')
)
.withColumn('"leg_range"', F.col('"max_leg_time"') - F.col('"min_leg_time"'))
)

我需要通过计算相同ID的第一个leg_sent和最后一个leg_received之间的差异并按源和目的地分组来计算全时差分。
谢谢!