如何在pyspark dataframe中获取列的唯一值并存储为新列

Question

基本上，我想知道某个客户在其他数据集中购买了多少品牌，并将其重命名为change brand，以下是我在Pandas中所做的

firstvalue=firstvalue.merge((pd.DataFrame(profile.groupby('msisdn')
                                          .handset_brand.nunique()
                                          .rename('hpbrand_change_num'))
                                          .reset_index()),how='left',on=['msisdn'])

下面是我在pyspark中所做的(没有合并)

fd_subsprofile.groupBy("msisdn")\
              .handset_brand.nunique()\
              .withColumn('hpbrand_change_num')\
              .reset_index()

错误消息

AttributeError: 'GroupedData' object has no attribute 'handset_brand'

然后，我试着

fd_subsprofile.groupBy("msisdn").select("handset_brand").count().show()

错误消息

AttributeError: 'GroupedData' object has no attribute 'select'

如何在pyspark中做到这一点？

Answer 1

同样的事情也可以在Pyspark中完成，如下所示-

nunique等价物- countDistinct、merge等价物- Join

import pyspark.sql.functions as F

profile_agg_sparkDF = profile.groupBy('id').agg(F.countDistinct(F.col('brand')).alias('change_brand'))

df = df.join(profile_agg_sparkDF
            ,df['id'] == profile_agg_sparkDF['id']
            ,'left'
        ).select(df['*'],profile_agg_sparkDF['change_brand'])