美文网首页
PySpark之聚合函数

PySpark之聚合函数

作者: HaloZhang | 来源:发表于2020-11-18 22:10 被阅读0次

    简介

    PySpark在DataFrame API中定义了内置的标准聚合(Aggregate)函数,当我们需要对DataFrame的列进行聚合操作时候,这些函数就可以派上用场。聚合函数是对一系列行进行相应操作,然后返回对这些行进行相应操作后的单一计算结果,即这些函数返回的一般是一个值。
    这些聚合函数一般接受以字符串形式的列类型或者列名称,以及一些函数特定的其他参数,并且它们的返回值是列的类型。

    聚合函数列表

    本文将介绍如下聚合函数:

    • approx_count_distinct
    • avg
    • collect_list
    • collect_set
    • countDistinct
    • count
    • first
    • last
    • kurtosis
    • max
    • min
    • mean
    • skewness
    • stddev
    • stddev_samp
    • stddev_pop
    • sum
    • sumDistinct
    • variance,var_samp, var_pop

    我们先创建一个基本的DataFrame结构,后续所有的聚合操作都是针对此数据集的。

    import pyspark
    from pyspark.sql import SparkSession
    from pyspark.sql.functions import approx_count_distinct,collect_list
    from pyspark.sql.functions import collect_set,sum,avg,max,countDistinct,count
    from pyspark.sql.functions import first, last, kurtosis, min, mean, skewness 
    from pyspark.sql.functions import stddev, stddev_samp, stddev_pop, sumDistinct
    from pyspark.sql.functions import variance,var_samp,  var_pop
    
    spark = SparkSession.builder.appName('SparkByExamples.com').getOrCreate()
    
    simpleData = [("James", "Sales", 3000),
        ("Michael", "Sales", 4600),
        ("Robert", "Sales", 4100),
        ("Maria", "Finance", 3000),
        ("James", "Sales", 3000),
        ("Scott", "Finance", 3300),
        ("Jen", "Finance", 3900),
        ("Jeff", "Marketing", 3000),
        ("Kumar", "Marketing", 2000),
        ("Saif", "Sales", 4100)
      ]
    schema = ["employee_name", "department", "salary"]
      
    df = spark.createDataFrame(data=simpleData, schema = schema)
    df.printSchema()
    df.show(truncate=False)
    
    结果:

    可以看到,数据主要有三列,分别是员工姓名,部门,薪资。

    approx_count_distinct

    approx_count_distinct()函数主要是用于计算一个组某一列中不同项目的计数值,这里的组你可以认为是很多列。
    比如说我们要计算员工的薪资一共有多少种不同数据:

    a = df.select(approx_count_distinct('salary')).collect()[0][0]
    print("number of different salary: ",a)
    
    结果:

    同理,我们也可以看下员工一共来自多少个不同你部门:

    a = df.select(approx_count_distinct('department')).collect()[0][0]
    print("number of different salary: ",a)
    
    输出:

    avg (average)

    avg()函数主要是用来计算某一列的平均值,比如我们计算所有员工的薪酬平均值:

    print("average salary: ", str(df.select(avg("salary")).collect()[0][0]))
    
    输出:

    collect_list

    collect_list()函数用于从输入列中返回全部的数值。下面的例子是返回"salary"列中所有的值

    df.select(collect_list("salary")).show(truncate=False)
    
    输出:

    collect_set

    collect_set()函数也是从输入列中返回全部不重复的值,还是以"salary"列为例:

    df.select(collect_set("salary")).show(truncate=False)
    
    输出:

    countDistinct

    countDistinct()函数用于计算某些列中不同元素的个数,比如这里返回所有行中"department"和"salary"这两列中元素至少有一个不相同的元素的个数:

    df2 = df.select(countDistinct("department", "salary"))
    df2.show(truncate=False)
    print("Distinct Count of Department & Salary: "+str(df2.collect()[0][0]))
    
    输出:

    可能你还是没太明白,注意看下原始数据,其中有两个“James”的部门和薪资是一样的,还有"Robert"和"Saif"的部门和薪资也是一样的。故10个人里面,要去掉重复的2个,故最后结果是8。

    count

    count()用户返回指定的某一列中元素的个数,比如这里返回"salary"列中包含几个元素:

    print("salary count: ", df.select(count("salary")).collect()[0])
    
    输出:

    first

    first()函数用来返回指定列中的第一个元素,当”ignoreNulls“被设置成"True"的时候,它返回第一个非空的元素。

    df.select(first("salary")).show(truncate=False)
    
    输出:

    last

    last()函数用来返回指定列中的第一个元素,当”ignoreNulls“被设置成"True"的时候,它返回第一个非空的元素。

    df.select(last("salary")).show(truncate=False)
    
    输出:

    kurtosis

    kurtosis()函数用于返回一个组中的峰度。关于峰度的定义可参见维基百科

    df.select(kurtosis("salary")).show(truncate=False)
    
    输出:

    max

    max()函数主要返回一列中的最大值。

    df.select(max("salary")).show(truncate=False)
    
    结果:

    min

    min()函数主要返回一列中的最大值。

    df.select(min("salary")).show(truncate=False)
    
    输出:

    mean

    mean()函数主要返回某一列的平均值,也叫做Avg。

    df.select(mean("salary")).show(truncate=False)
    
    输出:

    skewness

    skewness()函数返回某一列的偏度,偏度的定义见百度百科

    df.select(skewness("salary")).show(truncate=False)
    
    输出: image.png

    stddev, stddev_samp and stddev_pop

    stddev()和stddev_samp()是一样的, 返回某一列的样本标准偏差。
    and stddev_pop()返回某一列中数据的总体标准偏差。

    df.select(stddev("salary"), stddev_samp("salary"), \
        stddev_pop("salary")).show(truncate=False)
    
    输出:

    sum

    sum() 函数返回某一列中全部值的总和。

    df.select(sum("salary")).show(truncate=False)
    
    输出:

    sumDistinct

    sumDistinct()函数返回列中所有不同值的总和。

    df.select(sumDistinct("salary")).show(truncate=False)
    
    输出:

    variance, var_samp, var_pop

    variance(), var_samp()是一样的,返回某一列中所有值的无偏方差。
    var_pop()返回某一列中所有值的总体方差。

    df.select(variance("salary"),var_samp("salary"),var_pop("salary")) \
      .show(truncate=False)
    
    输出:

    参考

    相关文章

      网友评论

          本文标题:PySpark之聚合函数

          本文链接:https://www.haomeiwen.com/subject/lihhiktx.html