Python技术栈与Spark交叉数据分析双向整合进阶实战--大数据ML样本集案例实战

版权声明：本套技术专栏是做者（秦凯新）平时工做的总结和升华，经过从真实商业环境抽取案例进行总结和分享，并给出商业应用的调优建议和集群环境容量规划等内容，请持续关注本套博客。QQ邮箱地址：1120746959@qq.com，若有任何学术交流，可随时联系。

1 Spark SQL 分组聚合统计

• groupBy与agg搭配使用，agg(*exprs)，这个方法是GroupedData上用于计算聚合数据的方法，*exprs能够是一个 string到string组成的字典，key是须要聚合的列名，value是用于计算的聚合函数的名称。聚 合函数能够的取值有：avg, max, min, sum, count,mean。agg方法返回一个聚合后的 DataFrame对象。

  df = spark.read.csv('/sql/customers.csv',header=True)
    df.columns
    df.groupBy('Genre').agg({"Age":"mean","Annual Income (k$)":"max","Spending Score (1-100)":"count"}).show()
    
    +------+-----------------------------+-----------------------+------------------+
    | Genre|count(Spending Score (1-100))|max(Annual Income (k$))|          avg(Age)|
    +------+-----------------------------+-----------------------+------------------+
    |Female|                          112|                     99|38.098214285714285|
    |  Male|                           88|                     99| 39.80681818181818|
    +------+-----------------------------+-----------------------+------------------+
  复制代码

• 除了使用字符串以字典的方式指定，其实还能够使用聚合后的列的表达式来作相同的是，这 须要借助pyspark.sql.functions模块中的方法。

  from pyspark.sql.functions import *
    df = spark.read.csv('/sql/customers.csv',header=True)
    df.columns
    df.groupBy('Genre').agg(mean(df.Age)).show()
    
    +------+------------------+
    | Genre|          avg(Age)|
    +------+------------------+
    |Female|38.098214285714285|
    |  Male| 39.80681818181818|
    +------+------------------+
  复制代码

• apply(udf),使用pandas中的用户自定义函数做用在GroupedData的每一组数据之上，返 回结果做为一个DataFrame。udf用户自定义函数接收pandas.DataFrame做为参数，返回另 外一个pandas.DataFrame对象。这个方法是pyspark2.3中加入的新方法。经过@pandas_udf表示这是一个pandas的方法，参数为id long,v double,指定PandasUDFType 为分组map操做。(测试未经过)

  from pyspark.sql.functions import pandas_udf, PandasUDFType
    from pyspark.sql import SparkSession
    spark = SparkSession.builder.master('spark://hadoopmaste:7077').appName('apply').getOrCreate()
    df = spark.createDataFrame( [(1, 10.0), (1, 21.0), (2, 34.0), (2, 56.0), (2, 19.0)], ("id",
    "v"))
    @pandas_udf("id long, v double", PandasUDFType.GROUPED_MAP)
    def normalize(pdf):
        v = pdf.v
        print(type(v),type(pdf))
        return pdf.assign(v=(v - v.mean()) / v.std())
        
    df.groupby("id").apply(normalize).show()
    spark.stop()
    Pandas中DataFrame的assign方法是新建一个DataFrame而不会改变原来的DataFrame。
  复制代码

• avg(*cols)，给算给定的数值类型的列的平均值

  df = spark.read.csv('/sql/customers.csv',header=True)
    df.select(df.Age.cast('int').alias('age'),'Genre').groupBy('Genre').avg('age').show()
    
    +------+------------------+
    | Genre|          avg(age)|
    +------+------------------+
    |Female|38.098214285714285|
    |  Male| 39.80681818181818|
    +------+------------------+
  复制代码

• count(),返回每一个分组中数据的条数

  df = spark.read.csv('/sql/customers.csv',header=True)
    df.select(df.Age.cast('int').alias('age'),'Genre').groupBy('Genre').count().show()
  复制代码

• max(*cols),计算给定列中数值最大的值。

  df = spark.read.csv('/sql/customers.csv',header=True)
    df.select(df.Age.cast('int').alias('age'),'Genre','Annual Income
    (k$)').groupBy('Genre').max().show()
  复制代码

• mean(*cols)计算对应列的均值，列须要是数值类型

  spark.read.csv('/sql/customers.csv',header=True)
    df.select(df.Age.cast('int').alias('age'),'Genre','Annual Income
    (k$)').groupBy('Genre').mean().show()
  复制代码

• min(*cols) 计算对应列的最小值，列数值类型须要是数值类型

  spark.read.csv('/sql/customers.csv',header=True)
    df.select(df.Age.cast('int').alias('age'),'Genre','Annual Income
    (k$)').groupBy('Genre').min().show()
  复制代码

• sum(*cols)，计算指定列的和，列的类型须要是数值类型。

  spark.read.csv('/sql/customers.csv',header=True)
    df.select(df.Age.cast('int').alias('age'),'Genre','Annual Income (k$)').groupBy('Genre').sum().show()
  复制代码

未完待续

2 总结

Python技术栈与Spark交叉数据分析双向整合，让咱们在大数据融合分析达到了通用，能够发现Spark SQL 其实很大部分功能和Pandas雷同

秦凯新 于深圳 201812172352