使用来自另一个 Pyspark 数据帧的行信息过滤和求和一个 Pyspark 数据帧答案

【问题标题】：Filter and sum one Pyspark dataframe using row information from another Pyspark dataframe使用来自另一个 Pyspark 数据帧的行信息过滤和求和一个 Pyspark 数据帧
【发布时间】：2020-07-23 06:21:53
【问题描述】：

我有两个不同的数据框，称为 df 和 cr，每个数据框都有不同的列和行内容，如下所示。我正在尝试查找在 cr 数据框中显示的安装日期和移除日期之间给定机队类型在给定路线上发生的航班数量，并使用该总和创建一个新列。 df 数据框包含所需的车队、日期、路线和计数信息。我设想需要从每行 cr 中包含的信息中过滤掉 df。过滤将按车队、路线和包含的日期范围进行。 df 过滤后，路由计数将被求和并放入 cr 中给定行的新列，然后移动到下一行。我通常会在 python 中使用 for 循环来执行此操作，但我的数据帧非常大，并且在 Pyspark 中循环很乏味。

我目前的尝试：

def component_route_normalized(component_route_count, fleet_month_year_count):
    cr=component_route_count
    df=fleet_month_year_count

    temp=df.filter(
         F.col('month-year').between(pd.to_datetime(cr.date_installed),pd.to_datetime(cr.date_removed)) &
         F.col('route') == cr.route &
         F.col('fleet_type') == cr.fmis_fleet_type_code
    )

    cr=cr.withColumn('fleet_route_count', F.sum(temp.route_count))

return cr

cr 数据框内容示例：

+-----------+----------------------+------------------------+------------+-------+-------------+---------+-------+--------------------+--------------+------------+
|aircraft_id|nca_part_number_sse001|nca_serial_number_sse001|repair_cycle|  route|part__routing|departure|arrival|fmis_fleet_type_code|date_installed|date_removed|
+-----------+----------------------+------------------------+------------+-------+-------------+---------+-------+--------------------+--------------+------------+
|          1|        25-3246-9-0001|                     341|           8|PVG-EWR|           13|      PVG|    EWR|                 777|    2014-12-16|  2015-12-10|
|          1|        25-3246-9-0001|                     341|           8|EWR-TLV|           34|      EWR|    TLV|                 777|    2014-12-16|  2015-12-10|
|          1|        25-3246-9-0001|                     341|           8|CDG-EWR|            4|      CDG|    EWR|                 777|    2014-12-16|  2015-12-10|
|          1|        25-3246-9-0001|                     341|           8|DEL-EWR|           16|      DEL|    EWR|                 777|    2014-12-16|  2015-12-10|
|          1|        25-3246-9-0001|                     341|           8|EWR-MXP|            3|      EWR|    MXP|                 777|    2014-12-16|  2015-12-10|
|          1|        25-3246-9-0001|                     341|           8|EWR-LHR|            7|      EWR|    LHR|                 777|    2014-12-16|  2015-12-10|
|          1|        25-3246-9-0001|                     341|           8|TLV-EWR|           34|      TLV|    EWR|                 777|    2014-12-16|  2015-12-10|
|          1|        25-3246-9-0001|                     341|           8|NRT-IAH|           15|      NRT|    IAH|                 777|    2014-12-16|  2015-12-10|
|          1|        25-3246-9-0001|                     341|           8|IAH-FRA|            5|      IAH|    FRA|                 777|    2014-12-16|  2015-12-10|
|          1|        25-3246-9-0001|                     341|           8|EWR-CDG|            4|      EWR|    CDG|                 777|    2014-12-16|  2015-12-10|
|          1|        25-3246-9-0001|                     341|           8|BRU-EWR|            8|      BRU|    EWR|                 777|    2014-12-16|  2015-12-10|
|          1|        25-3246-9-0001|                     341|           8|NRT-EWR|           17|      NRT|    EWR|                 777|    2014-12-16|  2015-12-10|
|          1|        25-3246-9-0001|                     341|           8|FRA-EWR|           11|      FRA|    EWR|                 777|    2014-12-16|  2015-12-10|
|          1|        25-3246-9-0001|                     341|           8|EWR-PVG|           14|      EWR|    PVG|                 777|    2014-12-16|  2015-12-10|
+-----------+----------------------+------------------------+------------+-------+-------------+---------+-------+--------------------+--------------+------------+

df 数据框内容示例：

+----------+-------+----------+-----------+------------+-----------+
|month-year|  route|fleet_type|route_count|flight_month|flight_year|
+----------+-------+----------+-----------+------------+-----------+
|  6/1/2014|PHL-ORD|       737|         92|           6|       2014|
|  4/1/2014|IAH-TUL|       787|         23|           4|       2014|
|  4/1/2014|DFW-ORD|       737|         86|           4|       2014|
|  5/1/2014|BRO-IAH|       737|         33|           5|       2014|
|  4/1/2014|YQR-ORD|       787|          9|           4|       2014|
|  3/1/2014|SFO-IAH|       757|         58|           3|       2014|
|  4/1/2014|AUS-IAH|       BUS|         55|           4|       2014|
|  5/1/2014|AGU-IAH|       787|          1|           5|       2014|
+----------+-------+----------+-----------+------------+-----------+

【问题讨论】：

标签： python dataframe pyspark

【解决方案1】：

join 关于车队类型、日期范围和路线，然后是 groupBy 和 count 应该可以完成这项工作。您可以尝试这样的事情（可能需要调整以处理列类型）：

cond =  (F.col('cr.month-year').between(F.col('df.date_installed',F.col('df.date_removed')) &
         (F.col('df.route') == F.col('cr.route')) &
         (F.col('df.fleet_type') == F.col('cr.fmis_fleet_type_code')))

joined = cr.alias('cr').join(df.alias('df'), on=cond, how='inner')
counts = joined.groupBy(*cr.columns).agg(F.sum('route_count').alias('total_route_count'))

【讨论】：