Pyspark基础整理

1. 创建Spark用于读取数据，创建DataFrame

SparkSession是整个程序的入口，创建过程（还不懂Spark到底后台怎么整的，先写下来，后续再理解）

from pyspark.sql import SparkSession
spark = SparkSession.builder \
      .master("local") \
      .appName("Python Spark SQL basic example") \
      .config("spark.some.config.option", "some-value") \
      .getOrCreate()

各参数的意义：
builder：构造器，必须得有
master:
appName：当前Spark程序的名字，用于在网页UI显示，不设置会自动生成一个。可有可无。
config: 配置一些参数（不懂）
enableHiveSupport：增加支持 hive Support
getOrCreate：检查是否已经存在一个SparkSession，如果存在，配置当前设置到这个SparkSession，如果不存在，根据当前配置，创建一个新的SparkSession。

2. 读数据

读取某文件夹下的数据

color_df = spark.read.csv("/user/.../...",header=True)
data.persist()

3. 重命名

a. 创建的时候命名

data = spark.createDataFrame(data=[("Alberto", 2), ("Dakota", 2)],schema=['name','length'])

b. 重命名

# 
color_df2 = color_df.withColumnRenamed('color','color2')\
                    .withColumnRenamed('length','length2')

#
color_df.select(color_df.color.alias('color2')).show()

4. 查看字段类型

color_df.dtypes

5.选择列

# select
color_df.select('length').show()
color_df.select('length','color').show()
color_df.select(color_df.length).show()
# filter
# filter中有条件
color_df.filter(color_df.length>4).show()
color_df.filter(color_df.length>4)\
        .filter(color_df[0]!='white').show()
color_df.filter("color like 'b%'").show()
# spark.sql选择
# createOrReplaceTempView：创建临时视图，此视图的生命周期与用于创建此数据集的[SparkSession]相关联。

# DataFrame注册成SQL的表：
color_df.createOrReplaceTempView("color_df_tempView")
spark.sql('''select * from color_df_tempView''').show()

6.创建和删除列

from pyspark.sql.functions import udf, col,lit,concat_ws
# 创建列
# lit(0)这一列只有0
color_df.withColumn('newCol', lit(0)).show()
# 删除一列
color_df.drop('length').show()
# 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()

7.转json(后续学习怎么json转dataframe及保存)

color_df.toJSON().first()

8.排序

color_df.sort('length', 'color', ascending=False).show()
# 混合排序
color_df.sort(color_df.length.desc(), color_df.color.asc()).show()

9.删除含有缺失值的行及空值填充

df = spark_df.dropna()
# 或者
df=spark_df.na.drop()
# 用-1填充
spark_df =  spark_df.na.fill(-1)
df2.fillna(-1)
# 均值填充
# 计算缺失值，collect()函数将数据返回到driver端，为Row对象，[0]可以获取Row的值
mean_len = color_df.select(F.mean('length')).collect()[0][0]
df2 = spark_df.na.fill({'salary':mean_salary})
# 不同列用不同值进行填充
df2.na.fill({'salary':mean_salary, 'Dob':'unknown'})

# 如果'a'列缺失用'b'列填充
df.select(nanvl("a", "b").alias("r1"), nanvl(df.a, df.b).alias("r2")).show()

10. 分组统计

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

11.空值判断(数值类型的nan和一般类型None)

from pyspark.sql.functions import isnull, isnan
# 1.None 的空值判断
df = spark.createDataFrame([(1, None), (None, 2)], ("a", "b"))
# 判断'a'是否为空值,并将结构命名为'r1'和'r2'(isnull)
df.select(isnull("a").alias("r1"), isnull(df.a).alias("r2")).show()
# 2. nan判断(isnan)
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()

13.删除重复行(dropDuplicates())

df2 = df1.dropDuplicates()

14.类eval操作(expr):传入字符串,字符串被当作函数执行,返回执行结果

from pyspark.sql.functions import expr
# 返回'color'列的长度
color_df.select(expr('length(color)')).show()

15.返回行的最大值,最小值,F.greatest, F.least

from pyspark.sql import functions as F
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(F.greatest('emp_id','salary').alias('greatest'),
          F.least('emp_id','salary').alias('least')
          ).show()

16.取整:四舍五入(round),向下取整(bround)

# 四舍五入
from pyspark.sql.functions import round
df = spark.createDataFrame([(2.5,)], ['a'])

df.select(round('a', 0).alias('r')).show()

17.抽样

sample1 = color_df.sample(
    withReplacement=False, # 无放回抽样
    fraction=0.6,
    seed=1000)  
sample1.show()

18.描述型统计量和最值和方差

spark_df.describe().show()
spark_df.describe('a').show()
# 最大值和最小值
from pyspark.sql.functions import min, max
spark_df.select(min('d'), max('e')).show()
# 均值方差
from pyspark.sql.functions import mean, stddev  # 同样是在function里面

color_df.select(mean('uniform').alias('mean'),
                stddev('uniform').alias('stddev'))\
    .show()

19.查看列中唯一值distinct()

data.select('columns').distinct().show()