聚合功能在Spark中使用groupBy计数使用情况

  • Z时代
  • 2024-01-10
  • 分类：问答

我试图在pySpark的一行代码中进行多项操作，但不确定我的情况是否可行。

我的意图是不必将输出另存为新的数据框。

我当前的代码非常简单：

encodeUDF = udf(encode_time, StringType())
new_log_df.cache().withColumn('timePeriod', encodeUDF(col('START_TIME')))
  .groupBy('timePeriod')
  .agg(
    mean('DOWNSTREAM_SIZE').alias("Mean"),
    stddev('DOWNSTREAM_SIZE').alias("Stddev")
  )
  .show(20, False)

我的意图是count()在使用后添加groupBy，以得到与每个 列值匹配的记录计数，这些记录打印\显示为输出。

尝试使用时，出现groupBy(..).count().agg(..)异常。

.show（）打印，而无需将代码拆分为两行命令，例如：

new_log_df.withColumn(..).groupBy(..).count()
new_log_df.withColumn(..).groupBy(..).agg(..).show()

或更好的办法是将合并的输出输出到agg.show()-额外的一栏，用于说明与该行的值匹配的已记录记录数。例如：

timePeriod | Mean | Stddev | Num Of Records
    X      | 10   |   20   |    315

回答：

count()可以在内部使用，agg()因为groupBy表达式相同。

使用Python

import pyspark.sql.functions as func
new_log_df.cache().withColumn("timePeriod", encodeUDF(new_log_df["START_TIME"])) 
  .groupBy("timePeriod")
  .agg(
     func.mean("DOWNSTREAM_SIZE").alias("Mean"), 
     func.stddev("DOWNSTREAM_SIZE").alias("Stddev"),
     func.count(func.lit(1)).alias("Num Of Records")
   )
  .show(20, False)

pySpark

SQL函数文档

与斯卡拉

import org.apache.spark.sql.functions._ //for count()
new_log_df.cache().withColumn("timePeriod", encodeUDF(col("START_TIME"))) 
  .groupBy("timePeriod")
  .agg(
     mean("DOWNSTREAM_SIZE").alias("Mean"), 
     stddev("DOWNSTREAM_SIZE").alias("Stddev"),
     count(lit(1)).alias("Num Of Records")
   )
  .show(20, false)

count(1) 将按等于 count("timePeriod")

使用Java

import static org.apache.spark.sql.functions.*;
new_log_df.cache().withColumn("timePeriod", encodeUDF(col("START_TIME"))) 
  .groupBy("timePeriod")
  .agg(
     mean("DOWNSTREAM_SIZE").alias("Mean"), 
     stddev("DOWNSTREAM_SIZE").alias("Stddev"),
     count(lit(1)).alias("Num Of Records")
   )
  .show(20, false)

以上是 聚合功能在Spark中使用groupBy计数使用情况 的全部内容， 来源链接： utcz.com/qa/431082.html

回到顶部