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 F

def 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次的电影的平均分排名前10
    df.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的一个电影综合案例。