SparkSQL中的反連接后聯合

我正在運行 PySpark 腳本，其中我正在執行 2 個資料幀的反連接和聯合。但我想在 Spark SQL 中做到這一點。

df_src：

 ------- ------- 
|call_id|call_nm|
 ------- ------- 
|    100|     QC|
|    105|     XY|
|    110|     NM|
|    115|     AB|
 ------- ------- 

df_lkp：

 ------- ------- 
|call_id|call_nm|
 ------- ------- 
|    100|     QC|
|    105|     XY|
|    106|     XZ|
 ------- ------- 

我們有兩個資料框：df_src 和 df_lkp。我正在從 df_src 中提取不匹配的記錄：

df_unmatched = df_src.join(df_lkp, on=column_nm, how='left_anti')

它給出了這個結果：

df_unmatched

 ------- ------- 
|call_id|call_nm|
 ------- ------- 
|    110|     NM|
|    115|     AB|
 ------- ------- 

但我想使用 Spark SQL 來完成這一部分。我創建了臨時視圖vw_df_src&vw_df_lkp并嘗試運行以下查詢，但沒有得到結果。

unmatched_sql = "SELECT * from vw_df_src where {0} in (select {0} from vw_df_src minus select {0} from vw_df_lkp)".format('call_id')
df_unmatched = sqlContext.sql(unmatched_sql)

我還在合并資料框并洗掉重復項。我正在使用以下代碼：

df_src1 = df_lkp.union(df_src)
df_src1.show(10)
df_src2 = df_src1.dropDuplicates(['call_id'])

df_src2：

 ------- ------- 
|call_id|call_nm|
 ------- ------- 
|    110|     NM|
|    100|     QC|
|    115|     AB|
|    106|     XZ|
|    105|     XY|
 ------- ------- 

我也希望在 Spark SQL 中完成此操作。

我正在使用以下代碼創建臨時視圖：

df_src = sqlContext.read.format('com.databricks.spark.csv').option("delimiter", '\001').options(header='true',inferSchema='false').load(src_file_nm)
df_src.createOrReplaceTempView('vw_df_src')
df_lkp = sqlContext.read.format('com.databricks.spark.csv').option("delimiter", '\001').options(header='true',inferSchema='false').load(lkp_file)
df_lkp.createOrReplaceTempView('vw_df_lkp')

uj5u.com熱心網友回復：

反連接

spark.sql(
"""select * from vw_df_src LEFT ANTI JOIN 

vw_df_lkp  ON

vw_df_src.call_nm= vw_df_lkp.call_nm """).show()


 ------- ------- 
|call_id|call_nm|
 ------- ------- 
|    115|     AB|
|    110|     NM|
 ------- ------- 

如果在未初始化為 sql TRY 的筆記本單元中運行

%sql 
select * from vw_df_src LEFT ANTI JOIN 

vw_df_lkp  ON

vw_df_src.call_nm= vw_df_lkp.call_nm 

聯盟

在pyspark中，聯合回傳重復項，您必須drop_duplicates() 或使用distinct(). 在 sql 中，union 消除了重復項。因此，以下將做。Spark 2.0.0 unionall()重新調整重復并且union是事情

spark.sql(
"""select * from vw_df_src

union

select * from vw_df_lkp""" ).show()

uj5u.com熱心網友回復：

預設：

df_src = spark.createDataFrame(
    [(100, 'QC'),
     (105, 'XY'),
     (110, 'NM'),
     (115, 'AB')],
    ['call_id', 'call_nm']
)
df_lkp = spark.createDataFrame(
    [(100, 'QC'),
     (105, 'XY'),
     (105, 'XY'),
     (106, 'XZ')],
    ['call_id', 'call_nm']
)
df_src.createOrReplaceTempView('vw_df_src')
df_lkp.createOrReplaceTempView('vw_df_lkp')

根據您的要求，（反加入 聯合）可以這樣做：

spark.sql(
    """
    select *
    from vw_df_src as a
    anti join vw_df_lkp b on a.call_nm=b.call_nm
    union (select * from vw_df_lkp)
    """
).show()
#  ------- ------- 
# |call_id|call_nm|
#  ------- ------- 
# |    110|     NM|
# |    115|     AB|
# |    100|     QC|
# |    105|     XY|
# |    106|     XZ|
#  ------- ------- 

但是，似乎不需要反連接：

spark.sql(
    """
    select * from vw_df_src
    union
    select * from vw_df_lkp
    """
).show()
#  ------- ------- 
# |call_id|call_nm|
#  ------- ------- 
# |    100|     QC|
# |    105|     XY|
# |    115|     AB|
# |    110|     NM|
# |    106|     XZ|
#  ------- ------- 

標籤：阿帕奇火花 pyspark apache-spark-sql 联盟 反加入

上一篇：如何分解在一行中有多個字典的PySpark列

下一篇：Sparkcsv檔案大小比pandas大2倍