[博学谷学习记录] 超强总结，用心分享|Pyspark_SQL3

Pyspark

注：大家觉得博客好的话，别忘了点赞收藏呀，本人每周都会更新关于人工智能和大数据相关的内容，内容多为原创，Python Java Scala SQL 代码，CV NLP 推荐系统等，Spark Flink Kafka Hbase Hive Flume等等~写的都是纯干货，各种顶会的论文解读，一起进步。
今天继续和大家分享一下Pyspark_SQL3
#博学谷IT学习技术支持

前言

今天继续分享Pyspark_SQL3。

一、 电影分析案例

• 需求1：查询用户平均分
• 需求2：查询电影平均分
• 需求3：查询大于平均分的电影数量
• 需求4：查询高分电影中(>3)打分次数最多的用户，并求出此人打的平均分
• 需求5：查询每个用户的平均打分，最低打分，最高打分
• 需求6：查询被评分超过100次的电影的平均分排名前10

from pyspark import SparkContext, SparkConf
from pyspark.sql import SparkSession
from pyspark.sql.types import *
import pyspark.sql.functions as Fdef method01():# 需求1：查询用户平均分df.select("userid", "score").groupBy("userid").agg(F.round(F.avg("score"), 2).alias("u_s_avg")).orderBy("u_s_avg", ascending=False).show()def method02():# 需求2：查询电影平均分df.select("movieid", "score").groupBy("movieid").agg(F.round(F.avg("score"), 2).alias("m_s_avg")).orderBy("m_s_avg", ascending=False).show()def method03():# 需求3：查询大于平均分的电影数量df_avg_score = df.select("score").agg(F.avg("score").alias("avg_score"))df_movie_avg_score = df.select("movieid", "score").groupBy("movieid").agg(F.avg("score").alias("movie_avg_score"))print(df_movie_avg_score.where(df_movie_avg_score["movie_avg_score"] > df_avg_score.first()["avg_score"]).count())def method04():# 需求4：查询高分电影中(>3)打分次数最多的用户，并求出此人打的平均分# 4.1高分电影df_hight_score_movie = df.groupBy("movieid").agg(F.avg("score").alias("m_s_avg")).where("m_s_avg>3").select("movieid")# 4.2高分电影中打分次数最多的用户df_hight_count_user = df_hight_score_movie.join(df, "movieid", "inner").groupBy("userid").agg(F.count("movieid").alias("u_m_count")).orderBy("u_m_count", ascending=False).limit(1)# 4.3此用户的平均分df.where(df["userid"] == df_hight_count_user.first()["userid"]) \.select("userid", "score").groupBy("userid").agg(F.avg("score").alias("hight_user_avg_score")).show()def method05():# 需求5：查询每个用户的平均打分，最低打分，最高打分df.select("userid", "score").groupBy("userid").agg(F.avg("score").alias("u_avg_score")).show()df.select("userid", "score").groupBy("userid").agg(F.max("score").alias("u_avg_score")).show()df.select("userid", "score").groupBy("userid").agg(F.min("score").alias("u_avg_score")).show()def method06():# 需求6：查询被评分超过100次的电影的平均分排名前10df.groupBy("movieid").agg(F.count("movieid").alias("m_count"),F.avg("score").alias("m_avg_score")).where("m_count>100").orderBy("m_avg_score", ascending=False).limit(10).show()if __name__ == '__main__':print("move example")spark = SparkSession.builder.appName("move example").master("local[*]").getOrCreate()schema = StructType().add("userid", StringType()).add("movieid", StringType()) \.add("score", IntegerType()).add("datestr", StringType())df = spark.read \.format("csv") \.option("sep", "\t") \.schema(schema=schema) \.load("file:///export/data/workspace/ky06_pyspark/_03_SparkSql/data/u.data")method01()method02()method03()method04()method05()method06()spark.stop()

总结

今天主要和大家分享了Pyspark_SQL的一个电影综合案例。