在 pyspark 中获得不同连接输出的最佳方法是什么？答案

【问题标题】：What is the best way to get different join output in pyspark?在 pyspark 中获得不同连接输出的最佳方法是什么？
【发布时间】：2022-01-12 12:42:47
【问题描述】：

我想知道哪个在 spark 中最有效地获得低于 4 帧

df1 - left_anti
df2 - left_semi
df3 - right_anti
df4 - right_semi

方法 1：（连接 - 1，过滤器 - 4）

merged_df = left_df.join(right_df, join_condition, how='full_outer')
df1 = merged_df.filter(sf.col('right_df.col1').isNull()).select('left_df.*')
df2 = merged_df.filter(sf.col('right_df.col1').isNotNull()).select('left_df.*')
df3 = merged_df.filter(sf.col('left_df.col1').isNull()).select('right_df.*')
df4 = merged_df.filter(sf.col('left_df.col1').isNotNull()).select('right_df.*')

方法 2：（连接 - 4，过滤器 - 0）

df1 = left_df.join(right_df, join_condition, how='left_anti')
df2 = left_df.join(right_df, join_condition, how='left_semi')
df3 = left_df.join(right_df, join_condition, how='right_anti')
df4 = left_df.join(right_df, join_condition, how='right_semi')

和

join_condition = (sf.col('left_df.col1') = sf.col('right_df.col1'))

上面提到的哪种机制是有效的？

参考：https://medium.com/bild-journal/pyspark-joins-explained-9c4fba124839

编辑将col1 视为两个数据帧中的主键列（即不可为空）。

【问题讨论】：

这将取决于您接下来打算做什么（采取什么行动）。截至目前，1 和 2 都只产生 4 个执行计划。
我想对所有 4 个执行相同的操作.. 假设写入 hdfs 位置

标签： apache-spark pyspark apache-spark-sql

【解决方案1】：

在评论效率之前，只想指出，一般来说两种场景下的df_n可能并不完全相同：

>>> df1 = spark.createDataFrame([{'id1': 0, 'val1': "a"},{'id1': 1, 'val1': "b"},{'id1': None, 'val1': "df1"}])
>>> df2 = spark.createDataFrame([{'id2': 1, 'val2': "d"},{'id2': 2, 'val2': "e"},{'id2': None, 'val2': "df2"}])

>>> df1.show()
+----+----+
| id1|val1|
+----+----+
|   0|   a|
|   1|   b|
|null| df1|
+----+----+

>>> df2.show()
+----+----+
| id2|val2|
+----+----+
|   1|   d|
|   2|   e|
|null| df2|
+----+----+

>>> df1.join(df2, col("id1") == col("id2"), how="full_outer").show()                                                                 
+----+----+----+----+
| id1|val1| id2|val2|
+----+----+----+----+
|   0|   a|null|null|
|null| df1|null|null|
|null|null|null| df2|
|   1|   b|   1|   d|
|null|null|   2|   e|
+----+----+----+----+

>>> df1.join(df2, col("id1") == col("id2"), how="full_outer").filter(col('id2').isNull()).select(df1["*"]).show()
+----+----+
| id1|val1|
+----+----+
|   0|   a|
|null| df1|
|null|null|
+----+----+

>>> df1.join(df2, col("id1") == col("id2"), how="left_anti").show()
+----+----+
| id1|val1|
+----+----+
|   0|   a|
|null| df1|
+----+----+

>>> df1.join(df2, col('id1') == col('id2'), how='full_outer').filter(col('id2').isNotNull()).select(df1['*']).show()
+----+----+
| id1|val1|
+----+----+
|   1|   b|
|null|null|
+----+----+

>>> df1.join(df2, col('id1') == col('id2'), how='left_semi').show()
+---+----+
|id1|val1|
+---+----+
|  1|   b|
+---+----+

这当然是因为nulls 是如何被SQL 连接处理的，并且因为'full_outer' 连接的结果将包含双方所有不匹配的行。后者意味着用于创建df2 ("semi-join") 的k2.isNotNull() 过滤器不会消除由与左侧任何内容不匹配的右手键生成的任何null 填充行-完全外连接的手侧。例如：

>>> df1 = spark.createDataFrame([{'k1': 0, 'v1': "a"},{'k1': 1, 'v1': "b"},{'k1': 2, 'v1': "c"}])                                           
>>> df2 = spark.createDataFrame([{'k2': 2, 'v2': "d"},{'k2': 3, 'v2': "e"},{'k2': 4, 'v2': "f"}])
>>> df1.join(df2, col('k1') == col('k2'), how="full_outer").filter(col('k2').isNotNull()).select(df1["*"]).show()                           
+----+----+                                                                     
|  k1|  v1|
+----+----+
|null|null|
|   2|   c|
|null|null|
+----+----+
>>> df1.join(df2, col('k1') == col('k2'), how="left_semi").show()
+---+---+                                                                       
| k1| v1|
+---+---+
|  2|  c|
+---+---+

【讨论】：

是的，这是一个有效的观点。因为空值不相等。但我对数据没有空值的情况很感兴趣
即使没有空键，半连接的结果也将不同于它们的“替代方案”。下次我还可以请您用“EDIT”或“UPDATE”或smth清楚地标记您问题中的更改。特别是当编辑后答案变得有些无关紧要时。
当然，将突出显示更新。你能指出两种方法返回不同结果的情况吗

【解决方案2】：

[发布我的答案，希望有经验的用户可以修改]

我会说没关系。 Spark 将重新组织这些操作以进行优化，因此如果最终结果相同，则 DAG（有向无环图）和执行计划将是相同的。

如果目标是性能，那么 1 加入会更方便，因为它可以利用加入广播（如果右边的 df 不太大并且可以分配在内存中）

【讨论】：