SparkSQL 电影评价数据分析

birdskyws

发布于 2018-09-12 15:50:59

1K0

发布于 2018-09-12 15:50:59

当我做了很多Spark Core练习，喜欢写map、reduce，后来又开始用SparkSQL ，感觉SQL比mapReduce简洁优雅很多。 SQL是我的短板，通过Spark SQL又练习了group by、join 、case when 等语法。

数据集介绍

These files contain 1,000,209 anonymous ratings of approximately 3,900 movies made by 6,040 MovieLens users who joined MovieLens in 2000.

2000年，100万条电影评价数据集，包括3900部电影和6040个用户。

users.dat 格式：UserID::Gender::Age::Occupation::Zip-code
movies.dat 格式：MovieID::Title::Genres
ratings.dat 格式：UserID::MovieID::Rating::Timestamp

一、创建Dataset和Dataframe

JavaRDD<User> userRDD = spark.read().textFile("data/ml-1m/users.dat")
//返回Dataset<String>对象
.javaRDD()
//返回JavaRDD<String>对象
.map(line->{
    String[] arr = line.split("::");
    return new User(Long.parseLong(arr[0]),
            arr[1],Integer.parseInt(arr[2]),
            Integer.parseInt(arr[3]),
            arr[4]);
});
//返回JavaRDD<User>对象
Dataset<Row> userDF= spark.createDataFrame(userRDD,User.class);
//返回Dataset<Row>对象

spark.read().textFile 生成Dataset<String>对象
javaRDD() 生成JavaRDD<String>对象
map(String->{}) 生成JavaRDD<Object>对象
spark.createDataFrame(RDD,Object.class)生成Dataset<Row>对象

二、Group by 分组

统计各职业男性、女性人数

userDF.groupBy("occupation","gender")
                .agg(count("gender"))
                .orderBy("occupation")
                .show();

+----------+------+-------------+
|occupation|gender|count(gender)|
+----------+------+-------------+
|         0|     M|          479|
|         0|     F|          232|
|         1|     F|          209|
|         1|     M|          319|
|         2|     M|          176|
|         2|     F|           91|
|         3|     F|          100|
|         3|     M|           73|
|         4|     F|          234|
|         4|     M|          525|
|         5|     M|           81|
|         5|     F|           31|
|         6|     F|          102|
|         6|     M|          134|
|         7|     F|          139|
|         7|     M|          540|
|         8|     M|           14|
|         8|     F|            3|
|         9|     F|           89|
|         9|     M|            3|
+----------+------+-------------+
only showing top 20 rows

三、临时表

Dataset调用createOrReplaceTempView生成临时表，session内有效。 spark.sql执行sqll操作，可以选择创建的临时表。

userDF.createOrReplaceTempView("user");
spark.sql("select occupation,gender,count(1) from user " +
    " group by occupation,gender order by occupation desc,gender asc").show();

四、case when 实现更好的输出格式

case column 生成新的column。 sum 加和。

spark.sql("select occupation," +
        " sum(CASE gender WHEN 'F' THEN 1 ELSE 0 END)Female," +
        " sum(CASE gender WHEN 'M' THEN 1 ELSE 0 END)Male " +
        "from user group by occupation order by occupation desc").show();

+----------+------+----+
|occupation|Female|Male|
+----------+------+----+
|        20|    78| 203|
|        19|    15|  57|
|        18|     4|  66|
|        17|    52| 450|
|        16|    51| 190|
|        15|    28| 116|
|        14|    79| 223|
|        13|    34| 108|
|        12|    50| 338|
|        11|    22| 107|
|        10|    66| 129|
|         9|    89|   3|
|         8|     3|  14|
|         7|   139| 540|
|         6|   102| 134|
|         5|    31|  81|
|         4|   234| 525|
|         3|   100|  73|
|         2|    91| 176|
|         1|   209| 319|
+----------+------+----+
only showing top 20 rows

五、join 连接操作

select from tableA join tableB on tableA.XX = tableB.xx on 后面是连接条件。用UserID连接用户表和评论表：

spark.sql("SELECT gender,age,count(*) FROM user JOIN rating " +
        "ON user.userID = rating.userID WHERE movieID=2116 GROUP BY gender,age").show();
userDF.join(ratingDF,"userID")
        .filter("movieID=2116")
        .groupBy("gender","age")
        .count().show();

Dataset用不同的列名连接：

        Dataset<Row> ratingMovie = ratingDF.join(
                movieDF,ratingDF.col("movieID").equalTo(movieDF.col("movieID")));

六、filter

Dataset的filter是条件过滤语句row->{return true/false}。

ratingMovie.filter((FilterFunction<Row>) row ->{
                    if(row.getString(5).indexOf("Phantasm")>=0)
                        return true;
                    else
                        return false;
                }
        )