基于Spark SQL对新闻网站项目案例分析

Love The Way You Lie 2022-12-28 01:34 143阅读 0赞

### 目录 ###

*   *  一、需求分析
     *  二、数据准备
     *   *  （1）数据格式
         *  （2）基于Java开发离线数据生成器
         *  （3）基于业务数据创建hive表
     *  三、数据模型
     *   *  （1）需求一：对网站的pv分析
         *  （2）需求二：对网站的uv分析
         *  （3）需求三：新用户注册比率
         *  （4）需求四：用户跳出率
         *  （5）需求五：版块热度排行榜
     *  四、实施过程
     *  五、总结

## 一、需求分析 ##

新闻网站需求：

*  **PV：** PV是Page View，是指一个页面被所有用户访问次数的总和，页面被访问一次就被记 录1次PV。
 *  **UV：** UV是User View，是指一个页面被多少个用户访问了，一个用户访问一次是1次UV，一 个用户访问多次还是1次UV 。
 *  **新用户注册比率：** 当天注册用户数 / 当天未注册用户数
 *  **用户跳出率：** IP只浏览了一个页面就离开网站的次数/网站总访问数（PV）。
 *  **版块热度排行榜：** 根据每个版块每天被访问的次数，做出一个排行榜。

数据处理流程：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center]

## 二、数据准备 ##

### （1）数据格式 ###

**网站日志格式 ：**

date,timestamp,userid,pageid,section,action

**日志字段说明:**

date: 日期，yyyy-MM-dd格式 
    timestamp: 时间戳 
    userid: 用户id 
    pageid: 页面id 
    section: 版块 
    action: 用户行为，两类，点击页面和注册

**数据展示：**

2020-12-12,1607830121221,14175,977,entertainment,view
    2020-12-12,1607830121222,85760,931,car,view
    2020-12-12,1607830121222,15859,146,funny,view
    2020-12-12,1607830121222,null,883,country,view
    2020-12-12,1607830121222,null,525,entertainment,view
    2020-12-12,1607830121222,92861,553,aviation,view
    2020-12-12,1607830121222,86456,932,carton,view
    2020-12-12,1607830121222,29777,664,fashion,view
    2020-12-12,1607830121222,47610,805,technology,view
    2020-12-12,1607830121222,87222,415,tv-show,view

### （2）基于Java开发离线数据生成器 ###

package com.kfk.spark.web_analysis_project;
    
    import java.io.FileNotFoundException;
    import java.io.FileOutputStream;
    import java.io.OutputStreamWriter;
    import java.io.PrintWriter;
    import java.text.SimpleDateFormat;
    import java.util.Calendar;
    import java.util.Date;
    import java.util.Random;
    
    /** * 离线数据生成器 * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/12/12 * @time : 7:51 下午 */
    public class DataGenerator { 
        public static void main(String[] args) { 
    
            StringBuffer stringBuffer = new StringBuffer();
    
            // 版块内容
            String[] sections = new String[] { "country", "international", "sport",
                    "entertainment", "movie", "carton",
                    "tv-show", "technology", "internet",
                    "car", "military", "funny",
                    "fashion", "aviation", "government"};
    
            Random random = new Random();
            int[] newOldUserArr = new int[]{ 1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
    
            // 生成日期，默认是昨天
            Calendar calendar = Calendar.getInstance();
            calendar.setTime(new Date());
            calendar.add(Calendar.DAY_OF_YEAR,-1);
            Date yesterday = calendar.getTime();
            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd");
            String date = simpleDateFormat.format(yesterday);
    
            // 生成10000条访问数据
            for (int i = 0; i < 10000; i++){ 
                // 生成时间戳
                long timestamp = System.currentTimeMillis();
    
                // 随机生成userid（默认10000注册用户，每天1/10的访客是未注册用户）
                Long userid = 0L;
                int newOldUser = newOldUserArr[random.nextInt(10)];
                if (newOldUser == 1){ 
                    userid = null;
                } else { 
                    userid = (long)random.nextInt(100000);
                }
    
                // 随机生成pageid，共1000个页面
                long pageid = random.nextInt(1000);
    
                // 随机生成板块
                String section = sections[random.nextInt(sections.length)];
    
                // 生成固定的行为，view
                String action = "view";
    
                stringBuffer.append(date).append(",")
                        .append(timestamp).append(",")
                        .append(userid).append(",")
                        .append(pageid).append(",")
                        .append(section).append(",")
                        .append(action).append("\n");
            }
    
            // 生成100条注册数据
            for (int i = 0; i < 100 ; i++ ){ 
                // 生成时间戳
                long timestamp = System.currentTimeMillis();
    
                // 新用户都是userid为null
                Long userid = null;
    
                // 生成随机pageid，都是null
                Long pageid = null;
    
                // 生成随机版块，都是null
                String section = null;
    
                // 生成固定的行为，view
                String action = "register";
    
                stringBuffer.append(date).append(",")
                        .append(timestamp).append(",")
                        .append(userid).append(",")
                        .append(pageid).append(",")
                        .append(section).append(",")
                        .append(action).append("\n");
            }
    
            PrintWriter printWriter = null;
    
            try { 
                printWriter = new PrintWriter(new OutputStreamWriter(new FileOutputStream("/Users/caizhengjie/Desktop/new_access_"+date+".log")));
                printWriter.write(stringBuffer.toString());
            } catch (FileNotFoundException e) { 
                e.printStackTrace();
            } finally { 
                if (printWriter != null){ 
                    printWriter.close();
                }
            }
        }
    }

自动生成日志文件，日期默认是昨天的日期

### （3）基于业务数据创建hive表 ###

**创建hive数据表**

CREATE TABLE IF NOT EXISTS hivespark.news_access (
    cdate string,
    ctimestamp string,
    userid string,
    pageid string,
    section string,
    action string
    )
    ROW FORMAT DELIMITED
    FIELDS TERMINATED BY ','
    STORED AS TEXTFILE;

**加载数据**

load data local inpath '/opt/datas/new_access_2020-12-12.log' into table news_access;

## 三、数据模型 ##

根据需求分析，对每个需求进行数据模型的演示

数据模型创建：

create table pvuv(
    userid varchar(20),
    pageid varchar(20),
    );

insert into pvuv(userid,pageid) values('A','001');
    insert into pvuv(userid,pageid) values('B','001');
    insert into pvuv(userid,pageid) values('A','001');
    insert into pvuv(userid,pageid) values('A','001');
    insert into pvuv(userid,pageid) values('B','002');
    insert into pvuv(userid,pageid) values('A','002');
    insert into pvuv(userid,pageid) values('C','003');
    insert into pvuv(userid,pageid) values('A','003');
    insert into pvuv(userid,pageid) values('D','004');
    insert into pvuv(userid,pageid) values('C','003');
    insert into pvuv(userid,pageid) values('A','002');
    insert into pvuv(userid,pageid) values('D','001');

mysql> select * from pvuv;

+--------+--------+
    | userid | pageid |
    +--------+--------+
    | A      | 001    |
    | B      | 001    |
    | A      | 001    |
    | A      | 001    |
    | B      | 002    |
    | A      | 002    |
    | C      | 003    |
    | A      | 003    |
    | D      | 004    |
    | C      | 003    |
    | A      | 002    |
    | D      | 001    |
    +--------+--------+

--------------------

### （1）需求一：对网站的pv分析 ###

解决方案：按照pageid分组，求出每个pageid的个数，即就是每个网站的pv数。

数据模型：

select pageid,count(pageid) pv_count from pvuv group by pageid;

运行结果：

+--------+----------+
    | pageid | pv_count |
    +--------+----------+
    | 001    |        5 |
    | 002    |        3 |
    | 003    |        3 |
    | 004    |        1 |
    +--------+----------+

### （2）需求二：对网站的uv分析 ###

解决方案：先求出每个pageid中每个userid访问的次数，再统计pageid的个数，即就是每个网站的uv数。

数据模型：

select pageid,count(userid) count from pvuv group by pageid,userid;

运行结果：

+--------+-------+
    | pageid | count |
    +--------+-------+
    | 001    |     3 |
    | 001    |     1 |
    | 002    |     1 |
    | 002    |     2 |
    | 003    |     2 |
    | 003    |     1 |
    | 004    |     1 |
    | 001    |     1 |
    +--------+-------+

select pageid,count(pageid) uv_count from (select pageid,count(userid) uv_count from pvuv group by pageid,userid) t group by pageid;

运行结果：

+--------+----------+
    | pageid | uv_count |
    +--------+----------+
    | 001    |        3 |
    | 002    |        2 |
    | 003    |        2 |
    | 004    |        1 |
    +--------+----------+

### （3）需求三：新用户注册比率 ###

解决方案：

*  1、先获取昨天所有访问行为中新用户的访问总数，即where date = yesterday and userid = ‘null’
 *  2、获取昨天总的注册用户数
 *  3、新用户的注册比例 = 当天注册用户数 / 当天新用户的访问总数

### （4）需求四：用户跳出率 ###

解决方案：

*  1、先获取各网站总访问数（PV）
 *  2、再获取每个网站只有userid浏览一次的次数
 *  3、求出用户跳出率，统计再一张表上（IP只浏览了一个页面就离开网站的次数 / 网站总访问数（PV））

数据模型：  
pv：

select pageid,count(pageid) pv_count from pvuv group by pageid;

每个网站只有userid浏览一次的次数：

select pageid,count(count) unique_count from (select pageid,count(userid) count from pvuv group by pageid,userid) t where count = 1 group by pageid;

+--------+--------------+
    | pageid | unique_count |
    +--------+--------------+
    | 001    |            2 |
    | 002    |            1 |
    | 003    |            1 |
    | 004    |            1 |
    +--------+--------------+

求出用户跳出率：每个网站只有userid浏览一次的次数 / 网站总访问数（PV）

+--------+----------+--------------+--------+
    | pageid | pv_count | unique_count | rate   |
    +--------+----------+--------------+--------+
    | 001    |        5 |            2 | 0.4000 |
    | 002    |        3 |            1 | 0.3333 |
    | 003    |        3 |            1 | 0.3333 |
    | 004    |        1 |            1 | 1.0000 |
    +--------+----------+--------------+--------+

### （5）需求五：版块热度排行榜 ###

解决方案：版块热度排行榜：根据每个版块每天被访问的次数，做出一个排行榜

类似前面的求pv，也很简单，这里就不做数据模型讲解了，可以直接看代码中。

## 四、实施过程 ##

将每一个需求写入到一个方法中，运行结果以注释的形式显示。

package com.kfk.spark.web_analysis_project;
    
    import org.apache.spark.sql.Dataset;
    import org.apache.spark.sql.Row;
    import org.apache.spark.sql.SparkSession;
    import scala.collection.mutable.StringBuilder;
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/12/12 * @time : 9:32 下午 */
    public class WebLogsAnalysisSpark { 
        public static void main(String[] args) { 
    
            SparkSession spark = SparkSession
                    .builder()
                    .appName("WebLogsAnalysis")
                    .master("local")
                    .config("spark.sql.warehouse.dir", "/Users/caizhengjie/Document/spark/spark-warehouse")
                    .enableHiveSupport()
                    .getOrCreate();
    
            String yesterday = Tools.getYesterday();
    
            // 需求一：对网站的pv分析
            Dataset<Row> pagePV = calculatePagePV(yesterday,spark);
    
            // 需求二：对网站的uv分析
            Dataset<Row> pageUV = calculatePageUV(yesterday,spark);
    
            // 需求三：新用户注册比率
            double regisRate = calculateNewUserRegis(yesterday,spark);
    
            // 需求四：用户跳出率
            Dataset<Row> leaveRate = calculateUserLeaveRate(yesterday,spark);
    
            // 需求五：版块热度排行榜
            Dataset<Row> sectionSort = calculateSectionSort(yesterday,spark);
    
            // 将分析出的结果表写入到mysql数据库中
            writeData(sectionSort,"sectionSort_info");
    
        }
    
        /** * 计算每天每个页面PV并且对其排序 * 排序的好处：排序后插入到mysql，Java web系统要查询每天pv top10的页面，直接查询mysql表limit10就可以 * 如果不排序的话，那么Java web系统就要做排序，这样返回会影响Java web系统的性能，影响用户访问的时间 * @param date * @param spark */
        public static Dataset<Row> calculatePagePV(String date, SparkSession spark){ 
    
            // 写法一
            String sql = "select cdate,pageid,pv_count from " +
                    "(select cdate,pageid,count(pageid) pv_count from hivespark.news_access " +
                    "where action = 'view' and cdate = '"+date+"' group by cdate,pageid) " +
                    "news_table order by pv_count desc limit 10 ";
    
            // 写法二
            StringBuilder stringBuilder = new StringBuilder();
            stringBuilder.append("select cdate,pageid,pv_count from ");
            stringBuilder.append(" ( ");
            stringBuilder.append(" select cdate,pageid,count(pageid) pv_count from hivespark.news_access ");
            stringBuilder.append(" where action = 'view' and cdate = '"+date+"' group by cdate,pageid");
            stringBuilder.append(" ) ");
            stringBuilder.append(" news_table order by pv_count desc limit 10 ");
    
            spark.sql(stringBuilder.toString()).show();
    
            return spark.sql(stringBuilder.toString());
    
            /** * +----------+------+--------+ * | cdate|pageid|pv_count| * +----------+------+--------+ * |2020-12-12| 462| 23| * |2020-12-12| 719| 21| * |2020-12-12| 943| 20| * |2020-12-12| 616| 20| * |2020-12-12| 623| 20| * |2020-12-12| 517| 19| * |2020-12-12| 9| 19| * |2020-12-12| 649| 19| * |2020-12-12| 467| 18| * |2020-12-12| 671| 18| * +----------+------+--------+ */
    
        }
    
        /** * 计算每天每个页面UV并且对其排序 * 1、 * @param date * @param spark */
        public static Dataset<Row> calculatePageUV(String date, SparkSession spark){ 
    
            // 写法一
            String sql = "select cdate,pageid,count(pageid) uv_count from " +
                    "(select cdate,pageid,count(userid) count from hivespark.news_access " +
                    "where action = 'view' and cdate = '"+date+"' group by cdate,pageid,userid) " +
                    "t group by cdate,pageid order by uv_count desc limit 10";
    
            // 写法二
            StringBuilder stringBuilder = new StringBuilder();
            stringBuilder.append("select cdate,pageid,count(pageid) uv_count from ");
            stringBuilder.append(" ( ");
            stringBuilder.append(" select cdate,pageid,count(userid) count from hivespark.news_access ");
            stringBuilder.append(" where action = 'view' and cdate = '"+date+"' group by cdate,pageid,userid ");
            stringBuilder.append(" ) ");
            stringBuilder.append(" t group by cdate,pageid order by uv_count desc limit 10 ");
    
            spark.sql(stringBuilder.toString()).show();
    
            return  spark.sql(stringBuilder.toString());
    
            /** * +----------+------+--------+ * | cdate|pageid|uv_count| * +----------+------+--------+ * |2020-12-12| 462| 23| * |2020-12-12| 719| 19| * |2020-12-12| 623| 19| * |2020-12-12| 585| 18| * |2020-12-12| 467| 18| * |2020-12-12| 9| 18| * |2020-12-12| 517| 18| * |2020-12-12| 732| 18| * |2020-12-12| 437| 18| * |2020-12-12| 473| 17| * +----------+------+--------+ */
    
        }
    
        /** * 计算每天新用户的注册比例 * * 1、先获取昨天所有访问行为中新用户的访问总数，即where date = yesterday and userid = 'null' * 2、获取昨天总的注册用户数 * 3、新用户的注册比例 = 当天注册用户数 / 当天新用户的访问总数 * @param date * @param spark */
        public static double calculateNewUserRegis(String date, SparkSession spark){ 
    
            // 昨天所有访问行为中新用户的访问总数（写法一）
            String allNewViewSql = "select count(1) allNewView from hivespark.news_access where action = 'view' and cdate = '"+date+"' and userid = 'null' ";
    
            // 写法二
            StringBuilder allNewViewBuilder = new StringBuilder();
            allNewViewBuilder.append("select count(1) allNewView from hivespark.news_access ");
            allNewViewBuilder.append(" where action = 'view' and cdate = '"+date+"' and userid = 'null' ");
    
            spark.sql(allNewViewSql).show();
    
            /** * +----------+ * |allNewView| * +----------+ * | 977| * +----------+ */
    
            // 昨天总的注册用户数
    
            String allNewRegisSql = "select count(1) allNewRegis from hivespark.news_access where action = 'register' and cdate = '"+date+"'";
    
            // 写法二
            StringBuilder allNewRegisBuilder = new StringBuilder();
            allNewRegisBuilder.append("select count(1) allNewRegis from hivespark.news_access ");
            allNewRegisBuilder.append(" where action = 'register' and cdate = '"+date+"' ");
    
            spark.sql(allNewRegisSql).show();
    
            /** * +-----------+ * |allNewRegis| * +-----------+ * | 100| * +-----------+ */
    
            // 新用户的注册比例
            long allNewView = (long) spark.sql(allNewViewSql).collectAsList().get(0).get(0);
            long allNewRegis = (long) spark.sql(allNewRegisSql).collectAsList().get(0).get(0);
    
            double rate = (double) allNewRegis / (double) allNewView;
    
            System.out.println(Tools.formatDouble(rate,2));
    
            return rate;
    
            /** * 0.1 */
        }
    
        /** * 用户跳出率：IP只浏览了一个页面就离开网站的次数/网站总访问数（PV） * 1、先获取各网站总访问数（PV） * 2、再获取每个网站只有userid浏览一次的次数 * 3、求出用户跳出率，统计再一张表上 * @param date * @param spark */
        public static Dataset<Row> calculateUserLeaveRate(String date, SparkSession spark){ 
    
            String sql = "select t1.cdate,t1.pageid,t1.pv_count,t2.unique_count,round((t2.unique_count / t1.pv_count),2) rate from " +
                    "(select cdate,pageid,count(pageid) pv_count from hivespark.news_access " +
                    "where action = 'view' and cdate = '"+date+"' group by cdate,pageid) t1 " +
                    "left join " +
                    "(select cdate,pageid,count(count) unique_count from " +
                    "(select cdate,pageid,count(userid) count from hivespark.news_access " +
                    "where action = 'view' and cdate = '"+date+"' group by cdate,pageid,userid)" +
                    "t where count = 1 group by cdate,pageid) t2 " +
                    "on t1.pageid = t2.pageid order by t2.unique_count desc limit 10";
    
            spark.sql(sql).show();
    
            return spark.sql(sql);
    
            /** * +----------+------+--------+------------+----+ * | cdate|pageid|pv_count|unique_count|rate| * +----------+------+--------+------------+----+ * |2020-12-12| 462| 23| 23| 1.0| * |2020-12-12| 732| 18| 18| 1.0| * |2020-12-12| 467| 18| 18| 1.0| * |2020-12-12| 437| 18| 18| 1.0| * |2020-12-12| 585| 18| 18| 1.0| * |2020-12-12| 623| 20| 18| 0.9| * |2020-12-12| 719| 21| 18|0.86| * |2020-12-12| 285| 17| 17| 1.0| * |2020-12-12| 9| 19| 17|0.89| * |2020-12-12| 96| 17| 17| 1.0| * +----------+------+--------+------------+----+ */
    
        }
    
        /** * 版块热度排行榜：根据每个版块每天被访问的次数，做出一个排行榜 * @param date * @param spark */
        public static Dataset<Row> calculateSectionSort(String date, SparkSession spark){ 
    
            // 写法一
            String sql = "select cdate,section,se_count from " +
                    "(select cdate,section,count(section) se_count from hivespark.news_access " +
                    "where action = 'view' and cdate = '"+date+"' group by cdate,section) " +
                    "t order by se_count desc";
    
            // 写法二
            StringBuilder stringBuilder = new StringBuilder();
            stringBuilder.append("select cdate,section,se_count from ");
            stringBuilder.append(" ( ");
            stringBuilder.append(" select cdate,section,count(section) se_count from hivespark.news_access ");
            stringBuilder.append(" where action = 'view' and cdate = '"+date+"' group by cdate,section ");
            stringBuilder.append(" ) ");
            stringBuilder.append(" t order by se_count desc ");
    
            spark.sql(stringBuilder.toString()).show();
    
            return spark.sql(stringBuilder.toString());
    
            /** * +----------+-------------+--------+ * | cdate| section|se_count| * +----------+-------------+--------+ * |2020-12-12| technology| 733| * |2020-12-12| fashion| 711| * |2020-12-12| funny| 687| * |2020-12-12|international| 684| * |2020-12-12| internet| 678| * |2020-12-12| carton| 678| * |2020-12-12| aviation| 676| * |2020-12-12| movie| 669| * |2020-12-12| government| 657| * |2020-12-12| sport| 646| * |2020-12-12| military| 644| * |2020-12-12| car| 644| * |2020-12-12| tv-show| 642| * |2020-12-12| country| 630| * |2020-12-12|entertainment| 621| * +----------+-------------+--------+ */
    
        }
    
        /** * 将分析出的结果数据写入到mysql数据库中 * @param dataset * @param tablename */
        public static void writeData(Dataset<Row> dataset,String tablename){ 
    
            dataset.write()
                    .format("jdbc")
                    .option("url", "jdbc:mysql://bigdata-pro-m04/spark")
                    .option("dbtable", tablename)
                    .option("user", "root")
                    .option("password", "199911")
                    .save();
    
        }
    }

将日期和小数点的处理写入到工具类Tools中：

package com.kfk.spark.web_analysis_project;
    
    import java.math.BigDecimal;
    import java.text.SimpleDateFormat;
    import java.util.Calendar;
    import java.util.Date;
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/12/12 * @time : 9:42 下午 */
    public class Tools { 
    
        /** * 获取yesterday时间：yyyy-MM-dd * @return */
        public static String getYesterday(){ 
            Calendar calendar = Calendar.getInstance();
            calendar.setTime(new Date());
            calendar.add(Calendar.DAY_OF_YEAR,-1);
            Date yesterday = calendar.getTime();
    
            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd");
    
            return simpleDateFormat.format(yesterday);
        }
    
        /** * 数据的小数点处理 * @param num * @param scale * @return */
        public static double formatDouble(double num, int scale){ 
            BigDecimal bigDecimal = new BigDecimal(num);
            return bigDecimal.setScale(scale,BigDecimal.ROUND_HALF_UP).doubleValue();
        }
    }

## 五、总结 ##

*  一般来说，在小公司中，可能就是将我们的spark作业使用linux的crontab进行调度，将作业Jar放 在一台安装了spark客户端的机器上，并编写对应的spark-submit shell脚本，在crontab中可以配 置，比如每天凌晨3点执行一次spark-submit shell脚本，提交一次spark作业。
 *  对于大公司来说，可能需要比较复杂的开源大数据作业调度平台，比如常用的azkaban、Oozie等， 但是那几个BAT公司都是自己开发的调度服务。
 *  对于spark作业计算出来的结果，我们一般都会写入到关系型数据库，比如MySQL。这样的话，我们 可以基于mysql，用java web技术开发一套系统平台，来使用图表的方式展示每次spark计算出来的 关键指标。

--------------------

以上内容仅供参考学习，如有侵权请联系我删除！  
如果这篇文章对您有帮助，左下角的大拇指就是对博主最大的鼓励。  
您的鼓励就是博主最大的动力！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/3ce0966e4c4442778c6004782a857595.png