在Spark 1.5.x版本，增加了一系列内置函数到DataFrame API中，并且实现了code-generation的优化。与普通的函数不同，DataFrame的函数并不会执行后立即返回一个结果值，而是返回一个Column对象，用于在并行作业中进行求值。Column可以用在DataFrame的操作之中，比如select，filter，groupBy等。函数的输入值，也可以是Column。

种类	函数
聚合函数	approxCountDistinct, avg, count, countDistinct, first, last, max, mean, min, sum, sumDistinct
集合函数	array_contains, explode, size, sort_array

scala版本：
1、countdistinct：

package com.spark.study.core

import org.apache.spark.sql.SQLContext
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.catalyst.expressions.CountDistinct
import org.apache.spark.sql.functions._

object infunction {
def main(args:Array[String]){
    val conf = new SparkConf()
                  .setAppName("loadandsave")
                  .setMaster("local")
    val sc = new SparkContext(conf)
    val sqlcontext = new SQLContext(sc)
    import sqlcontext.implicits._
    val useraccesslog = Array(
    "2018-05-18,00001",
    "2018-05-18,00001",
    "2018-05-18,00002",
    "2018-05-18,00001",
    "2018-05-19,00001",
    "2018-05-19,00002",
    "2018-05-19,00002").toSeq

    val rdd1 = sc.parallelize(useraccesslog, 5)
    val rowrdd = rdd1.map(x => Row(x.split(",")(0),x.split(",")(1)))
    val structtype = StructType(Array(StructField("date",StringType,true),StructField("userid",StringType,true)))
    val df1 = sqlcontext.createDataFrame(rowrdd, structtype)
    df1.groupBy("date").agg('date, countDistinct('userid)).map(x =>Row(x(1),x(2))).collect().foreach(row => println(row))
      }
}

2、sum：

package com.spark.study.core

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.DoubleType
import org.apache.spark.sql.functions._

object infunctionsum {
def main(args:Array[String]){
    val conf = new SparkConf()
                  .setAppName("loadandsave")
                  .setMaster("local")
    val sc = new SparkContext(conf)
    val sqlcontext = new SQLContext(sc)
    import sqlcontext.implicits._
    val useraccesslog = Array(
    "2018-05-18,55.9,0001",
    "2018-05-18,33,0002",
    "2018-05-18,68,0001",
    "2018-05-18,115,0001",
    "2018-05-19,36,0002",
    "2018-05-19,67,0001",
    "2018-05-19,66,0002").toSeq

    val rdd1 = sc.parallelize(useraccesslog, 5)
    val filterrdd = rdd1.filter(x => if (x.split(",").length == 3) true else false)
    val rowrdd = rdd1.map(x => Row(x.split(",")(0),x.split(",")(1).toDouble,x.split(",")(2)))

    val structtype = StructType(Array(StructField("date",StringType,true),StructField("scales",DoubleType,true),StructField("userid",StringType,true)))
    val df1 = sqlcontext.createDataFrame(rowrdd, structtype)
    df1.groupBy("date").agg('date, sum('scales)).map(x =>Row(x(1),x(2))).collect().foreach(row => println(row))

}
}

ITPub博客

spark系列二：sparksql内置函数实例

最新文章