pyspark系列--datafrane进阶

Myth丶恋晨 2022-05-29 12:40 320阅读 0赞

**datafrane进阶**

*  1. 分组统计
 *  2. join 操作
 *  3. 缺失值处理
 *  4. 空值判断
 *  5. 缺失值处理
 *  6. 离群点
 *  7. 重复值
 *  8. 生成新列
 *  9. 类eval操作
 *  10. 行的最大最小值
 *  11. when操作
 *  12. lag,lead平移

# 1. 分组统计 #

分组统计应该是用的最多的方法了，比如分地区求平均值，最大最小值等。

# 分组计算1
    color_df.groupBy('length').count().show()
    
    # 分组计算2：应用多函数
    import pyspark.sql.functions as func
    color_df.groupBy("color").agg(func.max("length"), func.sum("length")).show()

# 2. join 操作 #

# 1.生成测试数据
    employees = [(1, "John", 25), (2, "Ray", 35), (3,"Mike", 24), (4, "Jane", 28), (5, "Kevin", 26), 
                 (6, "Vincent", 35), (7,"James", 38), (8, "Shane", 32), (9, "Larry", 29), (10, "Kimberly", 29),
                 (11, "Alex", 28), (12, "Garry", 25), (13, "Max",31)]
    employees=spark.createDataFrame(employees, schema=["emp_id","name","age"])
    employees.show()
    
    salary=[(1,1000),(2,2000),(3,3000),(4,4000)]
    salary=spark.createDataFrame(salary, schema=["emp_id","salary"])
    salary.show()
    
    department=[(1,1000),(2,2000),(3,3000),(4,4000)]
    department=spark.createDataFrame(department, schema=["emp_id","departement"])
    department.show()
    
    # 2.连接
    # join默认是内连接，最终结果会存在重复列名
    # 如果是pandas,重复列会用_x,_y等后缀标识出来，但spark不会
    # join会在最后的dataframe中存在重复列
    
    final_data = employees.join(salary, employees.emp_id == salary.emp_id, how='left')\
        .join(department, employees.emp_id==department.emp_id)
    final_data.show()
    
    # 3.如果两边的关联字段名相同，也可以省去很多麻烦
    final_data = employees.join(salary, on='emp_id', how='left')\
        .join(department, on='emp_id', how='left')
    final_data.show()
    
    # 4.多字段关联
    final_data = employees.join(salary, on=['emp_id','emp_name'], how='left')
    
    # 5.外连接
    # full outer join全外连接 
    df.join(df2, "name", "outer").select("id", "name", "age", "height").orderBy("id").show()

# 3. 缺失值处理 #

在join操作中，我们得到一个有缺失值的dataframe，接下来将对这个带有缺失值的dataframe进行操作

# 1.删除有缺失值的行
    clean_data=final_data.na.drop()
    clean_data.show()
    
    # 2.用均值替换缺失值
    import math
    from pyspark.sql import functions as func  # 导入spark内置函数
    # 计算缺失值，collect()函数将数据返回到driver端，为Row对象，[0]可以获取Row的值
    mean_salary = final_data.select(func.mean('salary')).collect()[0][0]
    clean_data = final_data.na.fill({
       'salary':mean_salary})
    
    # 3.如果一行至少2个缺失值才删除该行
    final_data.na.drop(thresh=2).show()
    
    # 4.填充缺失值
    # 对所有列用同一个值填充缺失值
    df1.na.fill('unknown').show()
    
    # 5.不同的列用不同的值填充
    df1.na.fill({
       'LastName':'--', 'Dob':'unknown'}).show()

# 4. 空值判断 #

有两种空值判断，一种是数值类型是nan，另一种是普通的None

# 类似 pandas.isnull
    
    from pyspark.sql.functions import isnull, isnan
    
    # 1.None 的空值判断
    df = spark.createDataFrame([(1, None), (None, 2)], ("a", "b"))
    df.select(isnull("a").alias("r1"), isnull(df.a).alias("r2")).show()
    
    # 2.nan的空值判断
    df = spark.createDataFrame([(1.0, float('nan')), (float('nan'), 2.0)], ("a", "b"))
    df.select(isnan("a").alias("r1"), isnan(df.a).alias("r2")).show()

# 5. 缺失值处理 #

如果col1为空则用col2填补，否则返回col1。  
类似 pandas 的 where 或者 combine\_first 方法

# pandas 
    #where即if-else函数
    np.where(isnull(a),b,a)
    
    # combine_first方法
    #如果a中值为空，就用b中的值填补
    a[:-2].combine_first(b[2:])
    
    #combine_first函数即对数据打补丁，用df2的数据填充df1中的缺失值
    df1.combine_first(df2)
    
    
    # pyspark
    df = spark.createDataFrame([(1.0, float('nan')), (float('nan'), 2.0)], ("a", "b"))
    df.select(nanvl("a", "b").alias("r1"), nanvl(df.a, df.b).alias("r2")).show()

# 6. 离群点 #

# 需要提醒的是，列的计算都是放在select里面的
    
    # 1.先计算均值
    mean_salary = final_data.select(func.mean('salary')).collect()[0][0]
    # 2.再计算方差
    devs=final_data.select(((final_data.salary-mean_salary)**2).alias('deviation'))
    # 3.再计算标准差
    stddev = math.floor(math.sqrt(devs.groupBy().avg('deviation').first()[0]))
    
    # 4.用均值的两倍标准差替代离群值
    no_outlier = final_data.select(
        final_data.emp_id, final_data.name, final_data.age, final_data.salary,
        func.when(final_data.salary.between(mean_salary-2*stddev, mean_salary+2*stddev), final_data.salary)
            .otherwise(mean_salary)
            .alias("updated_salary")
        )
    no_outlier.show()
    
    
    # func中有现成的常用统计函数，更加方便
    # 1.计算均值
    mean_salary = final_data.select(func.mean('salary')).collect()[0][0]
    # 2.计算标准差
    final_data.select(func.stddev('salary')).collect()[0][0]
    # 离群值替代就和上面的一致了

# 7. 重复值 #

# 重复值的处理，和pandas很像啊
    authors = [['Thomas','Hardy','June 2,1840'],
                ['Thomas','Hardy','June 2,1840'],
                ['Thomas','H',None],
                ['Jane','Austen','16 December 1775'],
                ['Emily',None,None]]
    
    df1 = spark.createDataFrame(authors,schema=["FirstName","LastName","Dob"])
    df1.show()
    
    # 删除重复值行
    df1.dropDuplicates().show()
    
    # 只要某一列有重复值，则去重
    df1.dropDuplicates(subset=['FirstName']).show()
    
    
    # pandas的方法
    df=pd.DataFrame(authors, columns=["FirstName","LastName","Dob"])
    df.drop_duplicates(subset=['FirstName'])

# 8. 生成新列 #

# 数据转换，可以理解成列与列的运算
    # 注意自定义函数的调用方式
    
    # 0.创建udf自定义函数，对于简单的lambda函数不需要指定返回值类型
    from pyspark.sql.functions import udf
    concat_func = udf(lambda name,age:name+'_'+str(age))
    
    # 1.应用自定义函数
    concat_df = final_data.withColumn("name_age", concat_func(final_data.name, final_data.age))
    concat_df.show()
    
    # 2.通过列生成另一列
    data_new=concat_df.withColumn("age_incremented",concat_df.age+1)
    data_new.show()
    
    # 3.某些列是自带一些常用的方法的
    df1.withColumn('Initial', df1.LastName.substr(1,1)).show()
    
    # 4.顺便增加一新列
    from pyspark.sql.functions import lit
    df1.withColumn('newCol', lit(0)).show()

# 9. 类eval操作 #

传入一个操作字符串，然后转成python代码执行，就像python的eval一样。

from pyspark.sql.functions import expr
    color_df.select(expr('length(color)')).show()

# 10. 行的最大最小值 #

求每一行的最大最小值，相当于`axis=1`

# 测试数据
    df=[(1,1000),(2,2000),(3,3000),(4,4000)]
    df=spark.createDataFrame(df, schema=["emp_id","salary"])
    df.show()
    
    # 求行的最大最小值
    from pyspark.sql.functions import greatest, least
    df.select(greatest('emp_id','salary').alias('greatest'),
              least('emp_id','salary').alias('least')
              ).show()
    # +--------+-----+
    # |greatest|least|
    # +--------+-----+
    # | 1000| 1|
    # | 2000| 2|
    # | 3000| 3|
    # | 4000| 4|
    # +--------+-----+

# 11. when操作 #

相当于SQL中的case when 操作

from pyspark.sql.functions import when
    
    # 1.case when age=2 then 3 else 4
    df.select(when(df['age'] == 2, 3).otherwise(4).alias("age"))show()
    
    # 2.case when age=2 when age=age+1 
    df.select(when(df.age == 2, df.age + 1).alias("age")).show()
    
    case when age<2 then age+2 else age end
    df.withColumn('age', when(df.age == 2, df.age + 1).otherwise(df2['age'])).show()

# 12. lag,lead平移 #

很好用的函数啊，特别是在处理时间序列的时候，和pandas的shift很像。

from pyspark.sql.functions import lag, lead
    df = spark.createDataFrame([(1, 2, 3) if i % 2 == 0 else (i, 2 * i, i % 4) for i in range(5)],
                               ["a", "b", "c"])
    df.show()
    
    df.select(lag('a', 1, 0).alias('lag')).show()
    df.select(lead('a', 1, 0).alias('lag')).show()