Spark SQL: 열 목록에 Aggregate 함수 적용

Spark SQL: 열 목록에 Aggregate 함수 적용

집합 함수를 데이터 프레임의 모든 열(또는 목록)에 적용할 수 있는 방법이 있습니까?groupBy? 즉, 모든 열에 대해 이 작업을 수행하지 않을 수 있는 방법이 있습니까?

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

집합 함수를 여러 열에 적용하는 방법은 여러 가지가 있습니다.

GroupedDataclass는 다음을 포함하여 가장 일반적인 함수들을 위한 많은 방법들을 제공합니다.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")

키와 기능을 값으로 사용하여 열 a인 사전/맵을 전달할 수도 있습니다.

• 파이썬

  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|
  ## +----+---------+
  

• 스칼라

  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|
  // +----+---------+------------------+---------+
  

마지막으로 다음과 같은 변수를 사용할 수 있습니다.

• 파이썬

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

• 스칼라

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

유사한 효과를 얻을 수 있는 다른 방법이 있지만, 이러한 방법은 대부분의 경우 충분합니다.

참고 항목:

• 스파크 데이터 프레임의 동일한 열에 대한 다중 Aggregate 작업

동일한 개념의 또 다른 예 - 그러나 두 개의 다른 열이 있다고 가정하면 - 그리고 당신은 각각 다른 agg 함수를 적용하려고 합니다. 즉, 당신은 그것들 각각에 다른 agg 함수를 적용하려고 합니다.

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

이 경우 별칭을 추가하는 방법은 아직 모르지만, 이를 달성하는 방법은 다음과 같습니다.

아래 예제 참조 - 지도 사용

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

집계를 만드는 방법에 대해서는 현재의 답변이 완벽하게 맞지만 질문에서 요청하는 열 별칭/이름 변경을 실제로 다루는 답변은 없습니다.

일반적으로 이 사건은 다음과 같이 처리합니다.

val dimensionFields = List("col1")
val metrics = List("col2", "col3", "col4")
val columnOfInterests = dimensions ++ metrics

val df = spark.read.table("some_table") 
    .select(columnOfInterests.map(c => col(c)):_*)
    .groupBy(dimensions.map(d => col(d)): _*)
    .agg(metrics.map( m => m -> "sum").toMap)
    .toDF(columnOfInterests:_*)    // that's the interesting part

마지막 줄은 기본적으로 집계된 데이터 프레임의 모든 열 이름을 원래 필드로 변경합니다.sum(col2)그리고.sum(col3)간단히 말하면col2그리고.col3.

pyspark의 경우, 다른 열 이름과 집성 함수를 결합할 수 있는 아래를 웰 이름 변경으로 사용할 수 있습니다.

from pyspark.sql.functions import expr
dims = ["col1"]
mertrics = ["col2", "col3", "col4"]
aggFuncs = ["sum", "avg", "max"]

aggs = [expr(f"{a}({c}) as {c}_{a}") for c in metrics for a in aggFuncs]
# or you can also do below, they are equivalent
# aggs = [expr(f"{a}({c})").alias(f"{c}_{a}") for c in metrics for a in aggFuncs]

df.groupBy(*dims).agg(*aggs).show()

언급URL : https://stackoverflow.com/questions/33882894/spark-sql-apply-aggregate-functions-to-a-list-of-columns