PySpark之選擇特徵select、篩選filter、聚合運算、group by、join table、inner join 、left join、right join、full outer join，如下所示：

from __future__ import print_function, division
from pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSession

## 啟動 Spark （如果你已經啟動就不需要）
spark = SparkSession.builder.master("local[2]").appName("test").enableHiveSupport().getOrCreate()
sc = spark.sparkContext

## 讀取資料集csv
df = spark.read.csv('../data/rating.csv', sep = ',', header = True) #自動判斷格式interSchema=True
df.show()

## 選擇特徵 ---select
#select userid from data   -- sql語句
df.select('userid').show() #篩選userid
#select userid ,movieid from data   -- sql語句
df.select('userid','movieid').show() 

## 對特徵進行操作 ---selectExpr
#select userid as id from data
df.selectExpr('userid as id').show()
#select movieid,rating * 2  as rating_2 from data
df.selectExpr('movieid', 'rating * 2 as rating_2').show()
df.printSchema()
df.selectExpr('cast(taring as DOUBLE)').printSchema() #轉換型別cast

## 篩選userid --- filter
#select * from data where rating > 3
df.filter('rating > 3').show()
#select * from data where userid = 2 and rating > 3
df.filter('userid == 2 and rating > 3').show()
#select userid, rating from data where userid = 2 and rating > 3
df.filter('userid == 2 and rating > 3').select('userid', 'rating').show()
df.select("userID", "rating").filter("userID = 2 and rating > 3").show()

## 聚合運算
#select count(*) from data
df.count()
df.agg({'userid':'count'}).show()
#select count(*) from data where userid = 1
df.filter('userid = 1').count()
#select count(userid) from data,select avg(rating) from data
df.agg({'userid':'count','rating':'avg'}).show()

## group by
##計算每個user評比了多少部電影，平均分數如何？
# select userid,count(*),avg(rating) from data group by userid
df.groupBy('userid').agg({'movieid':'count','rating':'avg'}).show()
from pyspark.sql.function import *
df.groupBy('userid').agg(count('movieid'), round(avg(df.rating), 2)).show()

## join table、inner join 、left join、right join、full outer join
# 建立資料框df_profile
d = [{'name': 'Alice', 'age': 1}, {'name': 'Bryan', 'age': 3}, {'name': 'Cool', 'age':2}]
df_profile = spark.createDataFrame(d) #轉換為資料框
df_profile.show()
# 建立資料框df_parents
d = [{'name': 'Jason', 'child': 'Alice'}, 
     {'name': 'Bill', 'child': 'Bryan'}, 
     {'name': 'Sera', 'child': 'Bryan'}, 
     {'name': 'Jill', 'child': 'Ken'}]
df_parents = spark.createDataFrame(d) #轉換為資料框
df_parents.show()
#inner join 
df_profile.join(df_parents, df_profile.name == df_parents.child).show()
#left join
df_profile.join(df_parents, df_profile.name == df_parents.child, 'left').show()
#right join
df_profile.join(df_parents, df_profile.name == df_parents.child, 'right').show()
#full outer join
df_profile.join(df_parents, df_profile.name == df_parents.child, 'outer').show()
---------------------