聚合函数在 Spark 中按组计数使用次数使用 Python与斯卡拉使用Java

java scala apache-spark pyspark apache-spark-sql

2022-08-31 16:37:08

我正在尝试在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)

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

尝试使用时，我会遇到异常。groupBy(..).count().agg(..)

有没有办法同时实现 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

答案 1

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 functions doc

与斯卡拉

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)

答案 2

聚合函数 在 Spark 中按组计数使用次数使用 Python与斯卡拉使用Java

使用 Python

与斯卡拉

使用Java

聚合函数在 Spark 中按组计数使用次数使用 Python与斯卡拉使用Java