您可以使用内置函数来获取聚合统计信息。以下是如何获得均值和标准偏差。
from pyspark.sql.functions import mean as _mean, stddev as _stddev, col
df_stats = df.select(
_mean(col('columnName')).alias('mean'),
_stddev(col('columnName')).alias('std')
).collect()
mean = df_stats[0]['mean']
std = df_stats[0]['std']
请注意，有三种不同的标准偏差功能。从文档中我使用的文档(stddev)返回以下内容：
聚合函数：返回无偏样本标准差
组中的表达
您也可以使用describe()方法：
df.describe().show()
更新：这是您处理嵌套数据的方法。
使用explode将值提取到单独的行中，然后调用mean和stddev，如上所示。
这是一个MWE：
from pyspark.sql.types import IntegerType
from pyspark.sql.functions import explode, col, udf, mean as _mean, stddev as _stddev
# mock up sample dataframe
df = sqlCtx.createDataFrame(
[(680, [[691,1], [692,5]]), (685, [[691,2], [692,2]]), (684, [[691,1], [692,3]])],
["product_PK", "products"]
)
# udf to get the "score" value - returns the item at index 1
get_score = udf(lambda x: x[1], IntegerType())
# explode column and get stats
df_stats = df.withColumn('exploded', explode(col('products')))\
.withColumn('score', get_score(col('exploded')))\
.select(
_mean(col('score')).alias('mean'),
_stddev(col('score')).alias('std')
)\
.collect()
mean = df_stats[0]['mean']
std = df_stats[0]['std']
print([mean, std])
哪个输出：
[2.3333333333333335, 1.505545305418162]
您可以使用numpy验证这些值是否正确：
vals = [1,5,2,2,1,3]
print([np.mean(vals), np.std(vals, ddof=1)])
说明：您的"products"列是list的{​​{1}}。调用list将为外部explode的每个元素创建一个新行。然后从每个展开的行中获取list值，这些行已定义为2元素"score"中的第二个元素。最后，在这个新列上调用聚合函数。

Python读取Excel文件取平均数
1.文件读取
下面我们将对这个excel文件进行读取


import xlrd                           #xlrd是excel文件读取库 只读写
data = xlrd.open_workbook('a.xlsx')   #打开excel文件
table = data.sheets()[0]              #打开第一张表格
nrows = table.nrows                   #获取表格的行数 

接下来利用For循环对每行进行处理，处理的目的是把想要的数据存储在列表当中
for i in range(nrows):
	name_.append(table.row_values(i)[0])
	data_.append(table.row_values(i)[2:5])

2.整合为字典
这一步是获取数据后第一个要处理的关键

初始化字典的方法有许多

但是最常用的方法还是使用

zip

可以把两个对应位置的列表进行整合为字典
dict_data = dict(zip(name_,data_))

输出结果
再根据字典输出为DataFrame
data  = pd.DataFrame(dict_data,index=['政治','纪律','品德','学习'])

再进行转置输出
data.T


2.平均数的读取
 data.mean(axis = 1)

mean（）里面axis表示的是坐标轴，axis = 0表示对纵向求平均值，axis = 1表示对横向求平均值

df = pd.DataFrame({'A': [1, 1, 1, 2, 2, 3, 3],'B':[1, 4, 3, 2, 1, 3, 3],'C':[1, 4, 3, 2, 1, 3, 3]}) 
df
>>>
    A	B	C
0	1	1	1
1	1	4	4
2	1	3	3
3	2	2	2
4	2	1	1
5	3	3	3
6	3	3	3

df['D'] = df[['A', 'B']].mean(axis=1)
df.loc[:, 'E'] = np.mean([df['A'], df['B']],0) #  如果没有0,求的是两列所有值的平均数
df
>>>
	A	B	C	  D	  E
0	1	1	1	1.0	1.0
1	1	4	4	2.5	2.5
2	1	3	3	2.0	2.0
3	2	2	2	2.0	2.0
4	2	1	1	1.5	1.5
5	3	3	3	3.0	3.0
6	3	3	3	3.0	3.0

SparkSql自定义聚合函数（弱类型DataFrame）求平均值
弱类型DataFrame提供了相关的聚合函数， 如 count()，countDistinct()，avg()，max()，min()；

除此之外，用户可以设定自己的自定义聚合函数。

代码测试前请确保各个组件均已安装

与强类型的代码思想差不多一样，但是书写有些变化，可以参考强类型写法：

强类型DataSet
1、环境准备
	1、准备json文件：
	{"name": "zhangsan","age": 20}
	{"name": "lisi","age": 30}
	{"name": "wangwu","age": 40}
	2、使用IDEA软件，创建maven工程
	3、添加pom依赖

2、maven工程pom依赖
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.spark.bigdata</groupId>
    <artifactId>sparkSql_Test</artifactId>
    <version>1.0-SNAPSHOT</version>
    <dependencies>
        <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-core -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.12</artifactId>
            <version>2.4.4</version>
        </dependency>
        
        <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-sql -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.12</artifactId>
            <version>2.4.4</version>
        </dependency>
    </dependencies>
</project>


3、创建类 AgeAvgFunction.Class
// 1)声明用户自定义聚合函数
// 2)继承UserDefinedAggregateFunction类
// 3)重写方法
class AgeAvgFunction extends UserDefinedAggregateFunction {
  // 函数输入的数据结构
  override def inputSchema: StructType = {
    new StructType().add("age", LongType)
  }

  // 计算时的数据结构（缓冲区的计算结构）
  override def bufferSchema: StructType = {
    new StructType().add("sum", LongType).add("count", LongType)
  }

  // 数据计算完毕之后的结构（函数返回数据类型）
  override def dataType: DataType = DoubleType

  // 稳定性
  override def deterministic: Boolean = true

  // 当前计算之前缓冲区的初始化（就是sum和count初始化是什么值）
  // 不考虑类型，只考虑结构
  override def initialize(buffer: MutableAggregationBuffer): Unit = {
    buffer(0) = 0L // 0代表第一个sum 初始化为0
    buffer(1) = 0L // 1代表第二个count 初始化为0
  }

  // 根据查询结构更新缓冲区数据
  override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
    buffer(0) = buffer.getLong(0) + input.getLong(0)
    buffer(1) = buffer.getLong(1) + 1
  }

  // 将多个节点的缓冲区合并
  override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
    // sum=缓冲区第一个位置的sum加上缓冲区第二个位置的sum
    buffer1(0) = buffer1.getLong(0) + buffer2.getLong(0)
    // count
    buffer1(1) = buffer1.getLong(1) + buffer2.getLong(1)
  }

  // 计算最终结果
  override def evaluate(buffer: Row): Any = {
    buffer.getLong(0).toDouble / buffer.getLong(1)
  }
}

4、创建SparkSql_aggregate.Scala
object SparkSql_aggregate {
  def main(args: Array[String]): Unit = {
    // 创建配置对象
    val conf = new SparkConf().setMaster("local[*]").setAppName("SparkSql_aggregate")
    // 创建SparkSql的环境对象
    val spark: SparkSession = new sql.SparkSession.Builder().config(conf).getOrCreate()
    /* 进行转换之前需要引入隐式转换规则
        这里Spark不是报名的含义，是SparkSession对象的名字*/
    import spark.implicits._
    // 创建聚合函数对象
    val udaf = new AgeAvgFunction
    // 注册聚合函数
    spark.udf.register("avgAge", udaf)
    // 使用聚合函数
    // 读取数据，构建DF
    val df: DataFrame = spark.read.json("input/sparksql.json")
    // 数据视图
    df.createOrReplaceTempView("user")
    // Sql语句
    spark.sql("select avgAge(age) from user").show()
    // 释放资源
    spark.stop()
  }

}

5、运行SparkSql_Aggregate.Scala
得到结果：