我有以下两个pySpark数据框架:
> df_lag_pre.columns
['date','sku','name','country','ccy_code','quantity','usd_price','usd_lag','lag_quantity']
> df_unmatched.columns
['alt_sku', 'alt_lag_quantity', 'country', 'ccy_code', 'name', 'usd_price']
现在我想在公共列上连接它们，所以我尝试以下操作:
> df_lag_pre.join(df_unmatched, on=['name','country','ccy_code','usd_price'])
，我得到以下错误信息:
AnalysisException: u'resolved attribute(s) price#3424 missing from country#3443,month#801,price#808,category#803,subcategory#804,page#805,date#280,link#809,name#806,quantity#807,ccy_code#3439,sku#3004,day#802 in operator !EvaluatePython PythonUDF#<lambda>(ccy_code#3439,price#3424), pythonUDF#811: string;'
在这个错误中显示的一些列，例如price，是构建df_lag的另一个数据框的一部分。我找不到任何关于如何解释这条消息的信息，所以任何帮助都将非常感谢。
您可以在pyspark中以这种方式执行join，请看看这对您是否有用:
df_lag_pre.alias("df1")
df_unmatched.alias("df2")
join_both = df1.join(df2, (col("df1.name") == col("df2.name")) & (col("df1.country") == col("df2.country")) & (col("df1.ccy_code") == col("df2.ccy_code")) & (col("df1.usd_price") == col("df2.usd_price")), 'inner')
更新:如果你得到colnot defined错误，请使用下面的import
from pyspark.sql.functions import col