如何解决Pyspark中的分析异常误差

Question

我在pyspark中有一个错误：

AnalysisException: u'Resolved attribute(s) week#5230 missing from
longitude#4976,address#4982,minute#4986,azimuth#4977,province#4979,
action_type#4972,user_id#4969,week#2548,month#4989,postcode#4983,location#4981 
in operator !Aggregate [user_id#4969, week#5230], [user_id#4969, 
week#5230, count(distinct day#4987) AS days_per_week#3605L]. 
Attribute(s) with the same name appear in the operation: week. 
Please check if the right attribute(s) are used

这似乎来自使用agg函数的代码段：

df_rs = df_n.groupBy('user_id', 'week')
            .agg(countDistinct('day').alias('days_per_week'))
            .where('days_per_week >= 1')
            .groupBy('user_id')
            .agg(count('week').alias('weeks_per_user'))
            .where('weeks_per_user >= 5').cache()

但是，我在这里看不到这个问题。而且我以前已经在相同数据上使用了这一行代码。

edit ：我一直在查看代码，错误的类型似乎来自这种加入：

df = df1.join(df2, 'user_id', 'inner')
df3 = df4.join(df1, 'user_id', 'left_anti).

，但仍未解决问题。

edit2 ：不幸的是，建议的问题与我的问题并不相似，因为这不是列名称歧义的问题，而是缺少属性的问题，在检查实际的dataframes时，这似乎并不丢失。<<<<<<<<<<<<<<<<<<<<<<<<<<<<<</p>

Answer 1

我遇到了同样的问题，并使用将解决的属性重命名为"降低"列在加入之前，对我进行了一些临时名称，这对我来说是一种解决方法，希望它对您也有帮助。不知道这个问题背后的真正原因，因为Spark 1.6 Spark-10925

Answer 2

我也多次面对这个问题，并遇到了这个问题在这里提到这是与火花相关的错误。根据本文，我提出了以下代码，该代码解决了我的问题。

代码可以处理左，右，内部和外部连接，尽管外部连接在此处以完整的外部工作。

def join_spark_dfs_sqlbased(sparkSession,left_table_sdf,right_table_sdf,common_join_cols_list=[],join_type="LEFT"):
    temp_join_afix="_tempjoincolrenames"
    join_type=join_type.upper()
    
    left=left_table_sdf.select(left_table_sdf.columns)
    right=right_table_sdf.select(right_table_sdf.columns)
    
    
    if len(common_join_cols_list)>0:
        common_join_cols_list=[col+temp_join_afix for col in common_join_cols_list]
    else:
        common_join_cols_list = list(set(left.columns).intersection(right.columns))
        common_join_cols_list=[col+temp_join_afix for col in common_join_cols_list]
        
    
    for col in left.columns:
        left = left.withColumnRenamed(col, col + temp_join_afix)
    left.createOrReplaceTempView('left')
     
    
    for col in right.columns:        
        right = right.withColumnRenamed(col, col + temp_join_afix)               
    right.createOrReplaceTempView('right')
                            
    
    
              
    non_common_cols_left_list=list(set(left.columns)-set(common_join_cols_list))
    non_common_cols_right_list=list(set(right.columns)-set(common_join_cols_list))
    
    
    unidentified_common_cols=list(set(non_common_cols_left_list)-set(non_common_cols_right_list))
    if join_type in ['LEFT','INNER','OUTER']:
        non_common_cols_right_list=list(set(non_common_cols_right_list)-set(unidentified_common_cols))
        common_join_cols_list_with_table=['a.'+col +' as '+col for col in common_join_cols_list]
    else:
        non_common_cols_left_list=list(set(non_common_cols_left_list)-set(unidentified_common_cols))
        common_join_cols_list_with_table=['b.'+col +' as '+col for col in common_join_cols_list]
                
    
    non_common_cols_left_list_with_table=['a.'+col +' as '+col for col in non_common_cols_left_list]        
    non_common_cols_right_list_with_table=['b.'+col +' as '+col for col in non_common_cols_right_list]
    
    non_common_cols_list_with_table=non_common_cols_left_list_with_table + non_common_cols_right_list_with_table    
    
    if join_type=="OUTER":
        join_type="FULL OUTER"
        
    join_type=join_type+" JOIN"
    select_cols=common_join_cols_list_with_table+non_common_cols_list_with_table
    
    common_join_cols_list_with_table_join_query=['a.'+col+ '='+'b.'+col for col in common_join_cols_list]
    query= "SELECT "+ ",".join(select_cols) + " FROM " + "left" + " a " + join_type + " " + "right" + " b" +" ON "+ " AND ".join(common_join_cols_list_with_table_join_query)
    
    print("query:",query)
    
    joined_sdf= sparkSession.sql(query)
        
    for col in joined_sdf.columns:     
        if temp_join_afix in col:   
            joined_sdf = joined_sdf.withColumnRenamed(col, col.replace(temp_join_afix,''))
            
    return joined_sdf