检查以下代码。

from pyspark.sql import functions as F

ts_tot_incv = spark.table("ts_tot_incv").where(F.col("mkt_offr_src_cd") == "A").select("mkt_offr_src_cd","offer_cd10","rp_offr_nm") # Fetching Data from ts_tot_incv & applying where condition. you can remove if don't want.

ts_tot_incv_5 = spark.table("ts_tot_incv_5").where(F.col("bus_unit") == "X").select("bus_unit","source_code".alias("compass_source_code"),"compass_offer_type") # Fetching data from ts_tot_incv_5 table & also applying where condition.

final_df = ts_tot_incv.join(ts_tot_incv_5,F.trim(ts_tot_incv["mkt_offr_src_cd"]) == F.trim(ts_tot_incv_5["source_code"]),"left_outer").select("mkt_offr_src_cd","offer_cd10","rp_offr_nm","bus_unit","compass_source_code","compass_offer_type").where((ts_tot_incv["mkt_offr_src_cd"] =="A") & (ts_tot_incv_5["bus_unit"] == "X")) # Applying left_outer join & selecting required columns & then applying where condition.

final_df.write.mode("append").saveAsTable("ts_tot") # This will create new table with ts_tot name & insert the data.