代码之家 › 专栏 › 技术社区 › Shankar Panda

Pyspark:如何在Pyspark中聚合列表中所有元素的数据?[副本]

pyspark-sql apache-spark-sql pyspark apache-spark

Shankar Panda · 技术社区 · 6 年前

我将所有字符串字段存储在列表对象中。然后,目前我正在传递for循环中的每个字段以计算聚合计数。

我正在寻找一种方法,一次获得所有字符串列的聚合计数。请帮忙。

Dataframe(输入数据)有这些记录

NoOfSegments,SegmentID,Country
3,2,Bangalore
3,2,Bangalore
3,3,Delhi
3,2,Delhi
3,3,Delhi
3,1,Pune
3,3,Bangalore
3,1,Pune
3,1,Delhi
3,3,Bangalore
3,1,Delhi
3,3,Bangalore
3,3,Pune
3,2,Delhi
3,3,Pune
3,2,Pune
3,2,Pune
3,3,Pune
3,1,Bangalore
3,1,Bangalore

我的代码:

        input_data.createOrReplaceTempView('input_data')

        sub="string"
        category_columns = [name for name, data_type in input_data.dtypes
                                if sub in data_type]
        df_final_schema = StructType([StructField("Country", StringType())
                           , StructField("SegmentID", IntegerType())
                           , StructField("total_cnt", IntegerType())
                        ])
        df_final=spark.createDataFrame([],df_final_schema)

        for cat_col in category_columns:
            query="SELECT {d_name} as Country,SegmentID ,(count(*) over(partition by {d_name},SegmentID)/ count(*) over(partition by NoOfSegments))*100 as total_cnt  from input_temp order by {d_name},SegmentID".format(d_name=cat_col)
            new_df=hc.sql(query)
            df_final = df_final.union(new_df)

结果:

1 回复 | 直到 6 年前

eliasah 6 年前

您可以使用 groupBy groupby ) :

from pyspark.sql import functions as F

total = df.select(F.sum("NoOfSegments")).take(1)[0][0]
df \
  .groupBy("SegmentID", "Country") \
  .agg(F.sum('NoOfSegments').alias('sums'))\
  .withColumn('total_cnt', 100 * F.col('sums')/ F.lit(total)) \
  .select('country', 'SegmentID', 'total_cnt') \
  .sort('country', 'SegmentID').show()
# +---------+---------+---------+
# |  Country|SegmentID|total_cnt|
# +---------+---------+---------+
# |Bangalore|        1|     10.0|
# |Bangalore|        2|     10.0|
# |Bangalore|        3|     15.0|
# |    Delhi|        1|     10.0|
# |    Delhi|        2|     10.0|
# |    Delhi|        3|     10.0|
# |     Pune|        1|     10.0|
# |     Pune|        2|     10.0|
# |     Pune|        3|     15.0|
# +---------+---------+---------+

推荐文章

user3579222 · 阅读以前的Spark API

5 月前

JFlo · 在PySpark笔记本中读取多个Parquet文件

5 月前

Matthew Thomas · partition覆盖动态和“逻辑”分区

9 月前

Jamal Khan · 如何在Apache Spark中读取500 GB的大文件CSV文件并对其执行聚合?

10 月前

Nakeuh · 从数组列中新建数据帧列

10 月前

maximodesousadias · 如何根据条件删除日期后的记录

10 月前

Ajay S Pal · 当调用函数时传递参数时,PySpark没有在函数内部创建Dataframe

1 年前

SUBHOJEET · 如何使用pyspark读取rds文件?

1 年前

Shankar Panda · 如何从org.apache.spark.sql获取密钥。在scala中键入列并将其放入列表变量中?

1 年前

Aaron Brazier · 连接2个pyspark数据帧并继续运行窗口sum和max

1 年前