Aggregatfunktion Zählen Sie die Nutzung mit groupBy in Spark

Question 1

Ich versuche, mehrere Operationen in einer Codezeile in pySpark durchzuführen, und bin mir nicht sicher, ob das für meinen Fall möglich ist.

Meine Absicht ist es nicht, die Ausgabe als neuen Datenrahmen zu speichern.

Mein aktueller Code ist ziemlich einfach:

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)

Und meine Absicht ist hinzuzufügen count() nach der Verwendung groupByum die Anzahl der Datensätze zu erhalten, die mit jedem Wert von übereinstimmen timePeriod Spalte, gedruckt\gezeigt als Ausgabe.

Beim Versuch zu verwenden groupBy(..).count().agg(..) Ich bekomme Ausnahmen.

Gibt es eine Möglichkeit beides zu erreichen count() und agg().show() druckt, ohne den Code in zwei Befehlszeilen aufzuteilen, zB:

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

Oder noch besser, um eine zusammengeführte Ausgabe zu erhalten agg.show() Ausgabe – Eine zusätzliche Spalte, die die gezählte Anzahl von Datensätzen angibt, die mit dem Wert der Zeile übereinstimmen. z.B:

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

Question 2

count() kann innen verwendet werden agg() wie groupBy Ausdruck ist gleich.

Mit 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-Funktionen doc

Mit Scala

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) zählt die Datensätze nach der ersten Spalte, die gleich ist count("timePeriod")

Mit 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)