加载 读取数据

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName('data_processing').getOrCreate()
df = spark.read_csv('data.csv',inferSchema=True,header=True) # inferSchema用于自行推断数据类型
-------------
df.columns #查看列名
(df.count,len(df.columns))#查看形状
df.printSchema() #类似info
df.show(5) #类似head
df.select('age','name').show()
df.describe().show()
-------------

添加一列新列

df = df.withColumn('age_after_10_years',(df['age']+10))

转换数据类型

from pyspark.sql.types import StringType,DoubleType
df['age'].cast(DoubleType())
-------------

筛选数据

df.filter(df['age']==20).show()
df.filter((df['age']==20) & (df['name']=='biob')).show()
求unique
df.select('name').distinct().show()
-------------

数据分组

df.groupBy('age').count().orderBy('age',ascending=False).show()
df.groupBy('age').mean() #返回 所有除去age的其他列的mean值
聚合
df.groupBy('age').agg({'salary':'sum'})
------------

自定义函数

from pyspark.sql.function import udf
def func():
    ***
    ***
    return **
UDF = udf(func,StringType())  #func的返回指的类型 可以是 StringType 也可以是 IntegerType() 看具体情况
df.withColumn('salary_new',UDF(df['salary']))  #新加一列 对df['salary'] 进行定义的函数操作
-----------
UDF = udf(lambda x:....,StringType())
df.withColumn('salary_new',UDF(df['salary']))
-----------

使用pandas_udf速度更快

from pyspark.sql.function import pandas_udf
def func():
    ...
UDF = pandas_udf(func,IntegerType())
df.withColumn(df['name'],UDF(df['name']))

函数的方法也可适用于多列

-----------

去重

df = df.dropDuplicates()
-----------

删除列

df = df.drop('salary')

写入数据 csv格式

path = 'result/train'
df.coalesce(1).write.format('csv').option('header','true').save(path)
-----------

写入数据 嵌套结构 适用于数据集特别大 可以对其进行压缩

path = 'result/train'
df.write.format('parquet').save(path)
------------

多列特征合并

from  pyspark.ml.linalg import Vector
from pyspark.ml.feature import VectorAssembler #VectorAssembler可将所有的特征汇成一列特征  

vec_assembler = VectorAssembler(inputCols=['city','career','age','gender'],outputCol='people_feature')
df = vec_assembler.transform(df)
-----------

划分数据集

train,test = df.randomSplit([0.8,0.2])
-----------
from pyspark.ml.regression import LinearRegression
linear_reg = LinearRegression(labelCol='label')
model = linear_reg.fit(train)
model_ = model.evaluate(train)
r2 = model_.r2 #还可以取更多的指标

res = model.evaluate(test).predictions
-----------

特征工程

from spark.ml.feature import StringIndexer  #类似于sklearn 的 label encoder
from spark.ml.feature import OneHotEncoder
from spark.ml.feature import VectorAssembler

df = StringIndexer(inputCol='city',outputCol='city_').fit_transform(df)
-----------

查看数据分布 这里用不了value_counts

df.groupBy('label').count().show()
-----------
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.classification import RandomForestClassifier

model = RandomForestClassifier(labelCol='label',numTrees=50).fit(train)
-----------

查看准确率 精度 auc

from spark.ml.evaluation import MulticlassClassificationEvaluator
from spark.ml.evaluation import BinaryClassificationEvaluator

predictions = model.transform(test)
accuracy = MulticlassClassificationEvaluator(labelCol='label',metricName='accuracy').evaluate(predictions) # metricName = 'weightedPrecision' 精度

auc = BinaryClassificationEvaluator(labelCol='label').evaluate(predictions)
----------

推荐算法  ALS(基于矩阵分解的一种方法)

from pyspark.ml.recommendation import ALS
rec = ALS(maxIter=10, regParam=0.01, userCol='userID', itemCol='itemID', ratingCol='rating', nonnegative=True, coldStartStrategy='drop')

其中 nonnegative 表示 不会在推荐系统中创建负数评分  coldStartStrategy可以防止生成任何nan评分预测
-----------
rec_model = rec.fit(train)
predicted_ratings = rec_model.transfrom(train)

from pyspark.ml.evaluation import RegressionEvaluator
evaluator = RegressionEvaluator(metricName='rmse',predictionCol='prediction',labelCol='rating')
rmase = evaluator.evaluate(predicted_ratings)

进行到这一步还没有完成推荐  只是完成了对测试集用户评分的预测 
-----------

接下来
推荐用户可能喜欢的排名靠前的电影 
先创建一系列独立的电影
unique_movies = df.select('movie').distinct()

假如我们对一位特定的userid 推荐电影  先过滤掉 他看过的电影
watched_movies = df.filter(df['userid']==userid).select('movie').distinct()
-----------

然后通过合并两张表 过滤空值 找出可以推荐的电影

total_movies = unique_movies.join(watched_movies,unique_movies.movie == watched_movies.movie)
remaining_movies = total_movies.where(col('watched_movies.movie').isNull()).select(unique.movies).distinct()
-----------

然后再用之前的模型对其进行评分预测 

recommendations = rec_model.transform(remaing_movies).orderBy('prediction',ascending=False)

之后还可以用 IndexToString 反变换  把推荐的电影数字 映射为 电影名字