SparkSQL：列のリストに集計関数を適用する

Question

グループを実行するときに、データフレームのすべての（またはリストの）列に集計関数を適用する方法はありますか？言い換えれば、すべての列でこれを避ける方法はありますか？

df.groupBy("col1") 
.agg(sum("col2").alias("col2"), sum("col3").alias("col3"), ...) 

ありがとうございます。

Answer 1

複数の列に集計関数を適用する方法は複数あります。

GroupedDataクラスは次のように直接使用することができるcount、max、min、meanとsum含むほとんどの一般的な機能、方法、の数を提供する：

• パイソン：

  df = sqlContext.createDataFrame(
      [(1.0, 0.3, 1.0), (1.0, 0.5, 0.0), (-1.0, 0.6, 0.5), (-1.0, 5.6, 0.2)], 
      ("col1", "col2", "col3")) 
  
  df.groupBy("col1").sum() 
  
  ## +----+---------+-----------------+---------+ 
  ## |col1|sum(col1)|  sum(col2)|sum(col3)| 
  ## +----+---------+-----------------+---------+ 
  ## | 1.0|  2.0|    0.8|  1.0| 
  ## |-1.0|  -2.0|6.199999999999999|  0.7| 
  ## +----+---------+-----------------+---------+ 
  

• スカラ

  val df = sc.parallelize(Seq(
      (1.0, 0.3, 1.0), (1.0, 0.5, 0.0), 
      (-1.0, 0.6, 0.5), (-1.0, 5.6, 0.2)) 
  ).toDF("col1", "col2", "col3") 
  
  df.groupBy($"col1").min().show 
  
  // +----+---------+---------+---------+ 
  // |col1|min(col1)|min(col2)|min(col3)| 
  // +----+---------+---------+---------+ 
  // | 1.0|  1.0|  0.3|  0.0| 
  // |-1.0|  -1.0|  0.6|  0.2| 
  // +----+---------+---------+---------+ 
  

必要に応じて、あなたはまた、値としての列のキーと機能を持つ辞書/マップを渡すことができ

df.groupBy("col1").sum("col2", "col3") 

を集約する必要がある列のリストを渡すことができます。

• Python

  exprs = {x: "sum" for x in df.columns} 
  df.groupBy("col1").agg(exprs).show() 
  
  ## +----+---------+ 
  ## |col1|avg(col3)| 
  ## +----+---------+ 
  ## | 1.0|  0.5| 
  ## |-1.0|  0.35| 
  ## +----+---------+ 
  

• Scalaの

  val exprs = df.columns.map((_ -> "mean")).toMap 
  df.groupBy($"col1").agg(exprs).show() 
  
  // +----+---------+------------------+---------+ 
  // |col1|avg(col1)|   avg(col2)|avg(col3)| 
  // +----+---------+------------------+---------+ 
  // | 1.0|  1.0|    0.4|  0.5| 
  // |-1.0|  -1.0|3.0999999999999996|  0.35| 
  // +----+---------+------------------+---------+ 
  

最後にあなたは可変引数を使用することができます。

• Pythonの

  from pyspark.sql.functions import min 
  
  exprs = [min(x) for x in df.columns] 
  df.groupBy("col1").agg(*exprs).show() 
  

• Scalaの

  import org.apache.spark.sql.functions.sum 
  
  val exprs = df.columns.map(sum(_)) 
  df.groupBy($"col1").agg(exprs.head, exprs.tail: _*) 
  

同様の効果を得るには他にもいくつかの方法がありますが、これはほとんどの場合十分です。

Answer 2

同じコンセプトのもう一つの例 - しかし、言う - あなたは2つの異なる列を持っている - ここではすなわち

f.groupBy("col1").agg(sum("col2").alias("col2"), avg("col3").alias("col3"), ...) 

とあなたがそれらのそれぞれに異なるAGG機能を適用したい、それを達成するための方法である - しかしIまだ

は、下記の例を参照してください、この場合にエイリアスを追加する方法がわからない - それはそう地図

val Claim1 = StructType(Seq(StructField("pid", StringType, true),StructField("diag1", StringType, true),StructField("diag2", StringType, true), StructField("allowed", IntegerType, true), StructField("allowed1", IntegerType, true))) 
val claimsData1 = Seq(("PID1", "diag1", "diag2", 100, 200), ("PID1", "diag2", "diag3", 300, 600), ("PID1", "diag1", "diag5", 340, 680), ("PID2", "diag3", "diag4", 245, 490), ("PID2", "diag2", "diag1", 124, 248)) 

val claimRDD1 = sc.parallelize(claimsData1) 
val claimRDDRow1 = claimRDD1.map(p => Row(p._1, p._2, p._3, p._4, p._5)) 
val claimRDD2DF1 = sqlContext.createDataFrame(claimRDDRow1, Claim1) 

val l = List("allowed", "allowed1") 
val exprs = l.map((_ -> "sum")).toMap 
claimRDD2DF1.groupBy("pid").agg(exprs) show false 
val exprs = Map("allowed" -> "sum", "allowed1" -> "avg") 

claimRDD2DF1.groupBy("pid").agg(exprs) show false