最近工作中用到pyspark, 在家自学整理了笔记
觉得有用的话&＃xff0c;点赞支持一下谢谢~
SparkContext & SparkSession
SparkContext主要用于创建和操作RDD
SparkSession实质上是SQLContext, HiveContext和SparkContext的组合。在Spark 2.x之前&＃xff0c;使用不同的功能&＃xff0c;需要引入不同的Context.
- 创建和操作RDD - SparkContext
- 使用streaming - StreamingContext
- 使用SQL - SQLContext
- 使用Hive - HiveContext
创建SparkSession
from pyspark.sql import SparkSession
spark &＃61; SparkSession.builder.getOrCreate()
数据表读写
&＃xff08;1&＃xff09;查看数据表
spark.catalog.listTables()
&＃xff08;2&＃xff09;从表中查询数据
# retrieve spark dataframe
query &＃61; "select * from demo"
data &＃61; spark.sql(query)
data.show()# spark dataframe to pandas dataframe
query &＃61; "select * from demo"
data &＃61; spark.sql(query).toPandas()
&＃xff08;3&＃xff09;输入数据
# csv to spark dataframe
data &＃61; spark.read.csv(file_path, header &＃61; True)
data.show()# read pandas dataframe to spark dataframe
# add spark_data to the catalog
spark_data &＃61; spark.createDataFrame(df)
spark_data.createOrReplaceTempView(&＃39;temp&＃39;)# read from catalog table
spark_data &＃61; spark.table("table")
数据表操作
spark dataframe是immutable, 因此每次返回的都是一个新的dataframe
&＃xff08;1&＃xff09;列操作
# add a new column
data &＃61; data.withColumn("newCol",df.oldCol&＃43;1)# replace the old column
data &＃61; data.withColumn("oldCol",newCol)# rename the column
data.withColumnRenamed("oldName","newName")# change column data type
data.withColumn("oldColumn", data.oldColumn.cast("integer"))
&＃xff08;2&＃xff09;条件筛选数据
# filter data by pass a string
temp1 &＃61; data.filter("col > 1000")# filter data by pass a column of boolean value
temp2 &＃61; data.filter(data.col > 1000)
&＃xff08;3&＃xff09;选择数据
# select based on column name
temp1 &＃61; data.select("col1","col2")
temp1 &＃61; data.select("col1*100 as newCol1")# select based on column object
temp2 &＃61; data.select(data.col1, data.col2)
temp2 &＃61; data.select(data.col1&＃43;1.alias(newCol1))
&＃xff08;4&＃xff09;聚合函数
# get the minimum value of a column
data.groupBy().min("col1")# group by on certain column and do calculation
data.groupBy("col1").max("col2")# agg function
import pyspark.sql.functions as F
data.groupBy("a","b").agg(F.stddev("c"))
&＃xff08;5&＃xff09;合并数据表
newData &＃61; data1.join(data2, on &＃61; "col", how &＃61; "leftouter")
newData &＃61; data1.join(data2, data1[&＃39;col1&＃39;] &＃61;&＃61; data2[&＃39;col2&＃39;])
京公网安备 11010802041100号