SparkSQL: apply aggregate functions to a list of column

SparkSQL: apply aggregate functions to a list of column

Asked on November 21, 2018 in Apache-spark.
Add Comment


  • 2 Answer(s)

    There are many ways for applying aggregate functions to multiple columns.

    In this GroupedData class provides a number of methods for the most similar functions, including count, max, min, mean and sum, which can be used directly as follows:

    Python:

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

    Scala

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

    Additionally we can pass a list of columns which should be aggregated

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

    Here we can also pass dictionary / map with columns a the keys and functions as the values:

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

    Atlast here varargs is used:

    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: _*)
    

    This could be more efficient method.

     

    Answered on November 21, 2018.
    Add Comment

    Here is a another instance of the same concept – but say –  If there are 2 different columns – and  want to apply different agg functions to each of them i.e

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

    This is the way used to achieve it

    By using Maps:

    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
    

     

    Answered on November 21, 2018.
    Add Comment


  • Your Answer

    By posting your answer, you agree to the privacy policy and terms of service.