兩次運行的輸入 df 結構相同，但輸出不同。只有第二次運行回傳所需的結果 ( df6)。我知道我可以為資料幀使用別名，這將回傳所需的結果。
問題。創建 的基礎 Spark 機制是df3什么？Spark 讀df1.c1 == df2.c2入了join'son子句，但很明顯它沒有注意提供的 dfs。引擎蓋下有什么？如何預測這種行為？
首次運行（df3結果錯誤）：
data = [
(1, 'bad', 'A'),
(4, 'ok', None)]
df1 = spark.createDataFrame(data, ['ID', 'Status', 'c1'])
df1 = df1.withColumn('c2', F.lit('A'))
df1.show()
# --- ------ ---- ---
#| ID|Status| c1| c2|
# --- ------ ---- ---
#| 1| bad| A| A|
#| 4| ok|null| A|
# --- ------ ---- ---
df2 = df1.filter((F.col('Status') == 'ok'))
df2.show()
# --- ------ ---- ---
#| ID|Status| c1| c2|
# --- ------ ---- ---
#| 4| ok|null| A|
# --- ------ ---- ---
df3 = df2.join(df1, (df1.c1 == df2.c2), 'full')
df3.show()
# ---- ------ ---- ---- ---- ------ ---- ----
#| ID|Status| c1| c2| ID|Status| c1| c2|
# ---- ------ ---- ---- ---- ------ ---- ----
#| 4| ok|null| A|null| null|null|null|
#|null| null|null|null| 1| bad| A| A|
#|null| null|null|null| 4| ok|null| A|
# ---- ------ ---- ---- ---- ------ ---- ----
第二次運行（正確df6結果）：
data = [
(1, 'bad', 'A', 'A'),
(4, 'ok', None, 'A')]
df4 = spark.createDataFrame(data, ['ID', 'Status', 'c1', 'c2'])
df4.show()
# --- ------ ---- ---
#| ID|Status| c1| c2|
# --- ------ ---- ---
#| 1| bad| A| A|
#| 4| ok|null| A|
# --- ------ ---- ---
df5 = spark.createDataFrame(data, ['ID', 'Status', 'c1', 'c2']).filter((F.col('Status') == 'ok'))
df5.show()
# --- ------ ---- ---
#| ID|Status| c1| c2|
# --- ------ ---- ---
#| 4| ok|null| A|
# --- ------ ---- ---
df6 = df5.join(df4, (df4.c1 == df5.c2), 'full')
df6.show()
# ---- ------ ---- ---- --- ------ ---- ---
#| ID|Status| c1| c2| ID|Status| c1| c2|
# ---- ------ ---- ---- --- ------ ---- ---
#|null| null|null|null| 4| ok|null| A|
#| 4| ok|null| A| 1| bad| A| A|
# ---- ------ ---- ---- --- ------ ---- ---
我可以看到物理計劃的不同之處在于內部使用了不同的連接（BroadcastNestedLoopJoin和SortMergeJoin）。但這本身并不能解釋為什么結果不同，因為對于不同的內部連接型別，它們應該仍然相同。
df3.explain()
== Physical Plan ==
BroadcastNestedLoopJoin BuildRight, FullOuter, (c1#23335 = A)
:- *(1) Project [ID#23333L, Status#23334, c1#23335, A AS c2#23339]
: - *(1) Filter (isnotnull(Status#23334) AND (Status#23334 = ok))
: - *(1) Scan ExistingRDD[ID#23333L,Status#23334,c1#23335]
- BroadcastExchange IdentityBroadcastMode, [id=#9250]
- *(2) Project [ID#23379L, Status#23380, c1#23381, A AS c2#23378]
- *(2) Scan ExistingRDD[ID#23379L,Status#23380,c1#23381]
df6.explain()
== Physical Plan ==
SortMergeJoin [c2#23459], [c1#23433], FullOuter
:- *(2) Sort [c2#23459 ASC NULLS FIRST], false, 0
: - Exchange hashpartitioning(c2#23459, 200), ENSURE_REQUIREMENTS, [id=#9347]
: - *(1) Filter (isnotnull(Status#23457) AND (Status#23457 = ok))
: - *(1) Scan ExistingRDD[ID#23456L,Status#23457,c1#23458,c2#23459]
- *(4) Sort [c1#23433 ASC NULLS FIRST], false, 0
- Exchange hashpartitioning(c1#23433, 200), ENSURE_REQUIREMENTS, [id=#9352]
- *(3) Scan ExistingRDD[ID#23431L,Status#23432,c1#23433,c2#23434]
uj5u.com熱心網友回復：
聯接取決于聯接資料幀的結構，但您構建這些資料幀的方式也會產生影響。如果您加入的兩個資料框共享相同的lineage，則您可能會在連接條件中遇到不明確的列問題，從而導致您在問題中描述的內容。
在您第一次運行時，當您df2從構建時df1，兩個資料幀共享相同的譜系。當您加入這兩個資料幀時，您實際上是在進行自聯接，Spark 選擇僅屬于已加入資料幀之一的錯誤列作為連接條件，導致笛卡爾積后跟始終為錯誤的過濾器。
在您的第二次運行中，由于兩個資料幀是獨立構建的，連接條件正確定義為兩列之間的相等性，每列屬于不同的資料幀。因此 Spark 執行經典連接。
詳細說明
正如pltc在他的回答中解釋的那樣，在您第一次運行時，Spark 不會為您的連接選擇正確的列。讓我們找出原因。
引擎蓋下是什么？
讓我們從獲取df1和df2使用explain. 這是物理計劃df1：
== Physical Plan ==
*(1) Project [ID#0L, Status#1, c1#2, A AS c2#6]
- *(1) Scan ExistingRDD[ID#0L,Status#1,c1#2]
這是物理計劃df2：
== Physical Plan ==
*(1) Project [ID#0L, Status#1, c1#2, A AS c2#6]
- *(1) Filter (isnotnull(Status#1) AND (Status#1 = ok))
- *(1) Scan ExistingRDD[ID#0L,Status#1,c1#2]
您可以通過在第一行開始看到(1) Project這兩個dataframesdf1并df2具有相同的列名和ID： [ID#0L, Status#1, c1#2, A AS c2#6]。這并不奇怪，因為df2從創建的df1，所以你可以看到df2作為df1額外的轉換。所以我們有以下參考：
df1.c1<=>df2.c1<=>c1#2df1.c2<=>df2.c2<=>A AS c2#6
當您加入df1and 時df2，意味著您進行了自加入。并且您的條件的以下所有組合都將被翻譯為c1#2 = A AS c2#6，這將為您留下簡化的連接條件c1#2 = A：
df1.c1 = df2.c2df1.c2 = df2.c1df2.c1 = df1.c2df2.c2 = df1.c1
當您在 Spark 中執行自聯接時，Spark 將重新生成正確資料幀的列 ID，以避免在最終資料幀中具有相同的列 ID。因此，在您的情況下，它將重寫df1. 所以列c1#2將參考列c1的df2。
現在您的條件不包含來自 的任何列df1，那么 Spark 將選擇執行笛卡爾積作為連接策略。由于兩個資料幀之一小到可以廣播，因此所選演算法將為BroadcastNestedLoopJoin。這是df3節目的物理計劃：
== Physical Plan ==
BroadcastNestedLoopJoin BuildRight, FullOuter, (c1#2 = A)
:- *(1) Project [ID#0L, Status#1, c1#2, A AS c2#6]
: - *(1) Filter (isnotnull(Status#1) AND (Status#1 = ok))
: - *(1) Scan ExistingRDD[ID#0L,Status#1,c1#2]
- BroadcastExchange IdentityBroadcastMode, [id=#75]
- *(2) Project [ID#46L, Status#47, c1#48, A AS c2#45]
- *(2) Scan ExistingRDD[ID#46L,Status#47,c1#48]
請注意， 的四個新列 IDdf1現在是[ID#46L, Status#47, c1#48, A AS c2#45]。
And when you execute this plan, as for the unique row of df2, the value of c1 is null which is different from A, thus join condition is always false. As you chose full outer join, you get the three rows (two from df1, one from df2) with null in columns coming from the other dataframe:
---- ------ ---- ---- ---- ------ ---- ----
| ID|Status| c1| c2| ID|Status| c1| c2|
---- ------ ---- ---- ---- ------ ---- ----
| 4| ok|null| A|null| null|null|null|
|null| null|null|null| 1| bad| A| A|
|null| null|null|null| 4| ok|null| A|
---- ------ ---- ---- ---- ------ ---- ----
Why for the second run I have the desired output?
For the second run, you create two independent dataframes. So if we look at the physical plan of df4 and df5, you can see that the column ids are different. Here is the physical plan of df4:
== Physical Plan ==
*(1) Scan ExistingRDD[ID#98L,Status#99,c1#100,c2#101]
And here is the physical plan of df5:
== Physical Plan ==
*(1) Filter (isnotnull(Status#124) AND (Status#124 = ok))
- *(1) Scan ExistingRDD[ID#123L,Status#124,c1#125,c2#126]
Your join condition is c1#100 = c2#126, c1#100 is c1 column from df4 and c2#126 is c2 column from df5. Each end of equality in join condition is from different dataframes, so Spark can perform the join as you expected.
Why this is not detected as Ambiguous Self Join?
Since Spark 3.0, Spark checks that the columns you're using for join are not ambiguous. If you inverted the order of df2 and df1 when joining them as follows:
df3 = df1.join(df2, (df1.c1 == df2.c2), 'full')
you would get the following error:
pyspark.sql.utils.AnalysisException: Column c2#6 are ambiguous.
So why don't we have this error when executing df2.join(df1, ...)?
You have your answer in the file DetectAmbiguousSelfJoin in Spark's code:
// When self-join happens, the analyzer asks the right side plan to generate
// attributes with new exprIds. If a plan of a Dataset outputs an attribute which
// is referred by a column reference, and this attribute has different exprId than
// the attribute of column reference, then the column reference is ambiguous, as it
// refers to a column that gets regenerated by self-join.
It means that when doing df2.join(df1, ...), we will only check columns used in join condition against df1. As in our case we didn't perform any transformation on df1, contrary to df2 that was filtered, exprIds of df1 columns didn't change and thus no ambiguous columns error is raised.
我在 Spark Jira 上創建了一個關于此行為的問題，請參閱SPARK-36874
如何預測這種行為？
您必須非常小心您的聯接是否是自聯接。如果你從一個資料幀開始df1，對它執行一些轉換來獲取df2，然后加入df1，df2你就有可能得到這種行為。為了減輕這一點，你應該始終把原來的資料幀的第一資料幀時做一個連接，所以有df1.join(df2, ...)代替df2.join(df1, ...)。通過這樣做，您將得到一個Analysis Exception: Column x are ambiguousif Spark 無法選擇正確的列。
uj5u.com熱心網友回復：
由于某種原因，Spark 無法正確區分您的列c1和c2列。這是獲得df3預期結果的修復程式：
df3 = df2.alias('df2').join(df1.alias('df1'), (F.col('df1.c1') == F.col('df2.c2')), 'full')
df3.show()
# Output
# ---- ------ ---- ---- --- ------ ---- ---
# | ID|Status| c1| c2| ID|Status| c1| c2|
# ---- ------ ---- ---- --- ------ ---- ---
# | 4| ok|null| A| 1| bad| A| A|
# |null| null|null|null| 4| ok|null| A|
# ---- ------ ---- ---- --- ------ ---- ---
轉載請註明出處，本文鏈接：https://www.uj5u.com/net/315258.html
標籤：阿帕奇火花 加入 火花 apache-spark-sql
上一篇：連接特定行值的列值