在 pySpark 上执行连接时“缺少已解决的属性”

Posted 2023-04-15

【中文标题】在 pySpark 上执行连接时“缺少已解决的属性”

【英文标题】："resolved attribute(s) missing" when performing join on pySpark

【发布时间】：2016-10-15 17:47:16

【问题描述】：

我有以下两个 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;'

出现此错误的某些列（例如价格）是构建 df_lag 的另一个数据框的一部分。我找不到有关如何解释此消息的任何信息，因此将不胜感激。

【问题讨论】：

!EvaluatePython PythonUDF 的来源是什么？你能提供一个最小的代码示例吗？

df_lag_pre 的血统似乎存在问题。如果您能提供完整的转换集，我们就能解决问题。

【参考方案1】：

你可以在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')

更新：如果你得到 col not defined 错误，请使用下面的导入

from pyspark.sql.functions import col

【讨论】：

您可以使用 : from pyspark.sql.functions import col 和 df1 是别名。无需定义且 df_lag_pre 和 df_unmatched 已经定义。希望这会有所帮助！