我有几个如下所示的数据框：

name_df:
| id | name |
| -- | ---- |
| 1  | Mark |
| 2  | Lisa |
| 2  | Josh |

age_df:
| name  | age |
| --    | --- |
| Mark  | 20  |
| John  | 25  |
| Lisa  | 35  |


prescription_df
| name  | prescription |
| --    | ------------ |
| Lisa  | True         |
| Mark  | False        |

因此，我在这里尝试猜测的是，是否有年龄超过 30 岁且处方设置为 True 的姓名。最后，如果满足这些条件，我必须返回 true 或 false。

我目前正在做两个不同的连接语句，如下所示：

df_1 = name_df.alias('main').join(
    age_df.alias('a'),
    (col('main.name') == col('a.name')) & col('a.age') > '30'
)
df_2 = df_1 alias('main').join(
    prescription_df.alias('b'),
    col('main.name') == col('b.name')) & col('b.prescription') == True
)

如果它是空的或者不是空的，我将返回 true 或 false ：

return False if df_2.isEmpty() else True

我想知道是否有“更好”的方法来做到这一点，或者更有效的方法来做到这一点。

解决方案

根据设计，PySpark 仅在您执行需要收集数据的操作时才执行您的任何代码。但在执行此操作之前，它会分析您之前的所有命令，并在提取数据之前创建优化的执行计划。您可以使用以下命令验证实际会发生什么.explain()

name_df.alias('main').join(
    age_df.alias('a'),
    (col('main.name') == col('a.name'))
).where(col('a.age') > 30).explain()

和

name_df.join(age_df.where(age_df.age > 30), on="name").explain()

和

name_df.join(age_df, on="name").where(age_df.age > 30).explain()

最终都会在运行时执行相同的代码

解释向你展示了物理计划

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- SortMergeJoin [name#613], [name#643], Inner
   :- Sort [name#613 ASC NULLS FIRST], false, 0
   :  +- Exchange hashpartitioning(name#613, 200), ENSURE_REQUIREMENTS, [plan_id=2294]
   :     +- Filter isnotnull(name#613)
   :        +- Scan ExistingRDD[name#613,id#614L]
   +- Sort [name#643 ASC NULLS FIRST], false, 0
      +- Exchange hashpartitioning(name#643, 200), ENSURE_REQUIREMENTS, [plan_id=2295]
         +- Filter ((isnotnull(age#644L) AND isnotnull(name#643

出于个人喜好，我可能会这样写，因为这样理解起来非常直观。

prescription_df.where(prescription_df.prescription == True).join(
    age_df.where(age_df.age > 30), on="name"
).join(name_df, on="name").isEmpty()