spark之淘宝app一个月数据流量运营分析

一.统计指标
针对淘宝app一个月的数据进行流量运营分析，主要包括：
              1.app流量分析
                  (1).pv页面浏览量
                  (2).uv独立访客
                  (3).访问深度
                  (4).每天访客数和成交量
                  (5).不同时段的访客数及成交量
                  (6).流失率
              2.用户行为分析
                  (1).用户浏览活跃时段
                  (2).用户购买活跃时段
                  (3).用户浏览最多的类目
                  (4).用户收藏最多的类目
                  (5).用户加购最多的类目
                  (6).用户购买最多的类目
                  (7).最近 30 天购买次数
                  (8).最近 7 天的活跃天数
                  (9).复购率
              3.商品分析
                  (1).浏览量最多和最少的 TOP5 类目
                  (2).收藏量最多和最少的 TOP5 类目
                  (3).加购量最多和最少的 TOP5 类目
                  (4).购买量最多和最少的 TOP5 类目

二.数据
 数据集【taobao_persona.csv】来自和鲸社区 https://www.kesci.com/home/dataset/5ef7024363975d002c9235d3，
 记录了 2014 年 11 月 18 日至 2014 年 12 月 18 日的电商 APP 部分用户行为记录，原始数据共 23291027 条，共有 6 个属性。

      属性名                 属性值
      user_id                 用户 ID,一个 ID 允许触发多个行为
      item_id                 商品 ID
      behavior_type           用户行为类型，包括浏览=1，收藏=2，加入购物车=3，购买=4
      user_geohash            用户地理位置
      item_category           商品所属类目的 ID
      time                    用户行为发生的日期和时刻，值域为 2014/11/18-2014/12/18

三.代码(spark3.0,java1.8)
详细代码见：TrafficOperationAnalysis淘宝APP一个月数据的流量运营分析(https://github.com/jiangnanboy/spark_tutorial)

 /**
     * 1.app流量分析
     * 2.用户行为分析
     * 3.商品分析
     *
     * 从 30 天里 APP 的总浏览量、总独立访客量、访问深度、每天访客量及成交量、每个时段访客数及成交量、各环节的流失率分析近 30 天的 APP 流量，及时发现流量运营异常。
     *
     * 从用户浏览活跃时段、用户购买活跃时段、用户浏览或收藏或加购或购买最多的类目、最近 30 天购买频次、最近 7 天活跃天数、复购率分析用户行为，了解用户喜好，以便进行后续的用户分类和商品推荐。
     *
     * 从商品类目的浏览量、收藏量、加购量、成交量分析不同商品类目的受欢迎程度，定位主要盈利和较为冷门的商品类目。
     *
     * 利用传统的 RFM 模型分析用户价值，由于原始数据缺少用户的消费金额，因此本项目只从 RF 划分用户群体。
     *
     * 利用 ARIMA 模型预测未来七天（2014/12/19-2014/12/25）的访客量，包括平稳性检验、差分法平稳序列、拟合预测。
     *
     * @param session
     */
    public static void analysis(SparkSession session) {
        String path = PropertiesReader.get("taobao_persona_csv");

        /**
         * 加载数据查看schema，样本数量为：23291027
         * +--------+---------+-------------+------------+-------------+-------------+
         * | user_id|  item_id|behavior_type|user_geohash|item_category|         time|
         * +--------+---------+-------------+------------+-------------+-------------+
         * |10001082|285259775|            1|     97lk14c|         4076|2014-12-08 18|
         * |10001082|  4368907|            1|        null|         5503|2014-12-12 12|
         * |10001082|  4368907|            1|        null|         5503|2014-12-12 12|
         * |10001082| 53616768|            1|        null|         9762|2014-12-02 15|
         * |10001082|151466952|            1|        null|         5232|2014-12-12 11|
         * +--------+---------+-------------+------------+-------------+-------------+
         *
         *  |-- user_id: integer (nullable = true)
         *  |-- item_id: integer (nullable = true)
         *  |-- behavior_type: integer (nullable = true)
         *  |-- user_geohash: string (nullable = true)
         *  |-- item_category: integer (nullable = true)
         *  |-- time: string (nullable = true)
         *
         */
        Dataset<Row> dataset = session.read()
                .option("sep", ",")
                .option("header", "true")
                .option("inferSchema", "true")
                .csv(path);

        /**
         * 将time拆为两列date和time
         * +--------+---------+-------------+------------+-------------+----+----------+
         * | user_id|  item_id|behavior_type|user_geohash|item_category|time|      date|
         * +--------+---------+-------------+------------+-------------+----+----------+
         * |10001082|285259775|            1|     97lk14c|         4076|  18|2014-12-08|
         * |10001082|  4368907|            1|        null|         5503|  12|2014-12-12|
         * |10001082|  4368907|            1|        null|         5503|  12|2014-12-12|
         * |10001082| 53616768|            1|        null|         9762|  15|2014-12-02|
         * |10001082|151466952|            1|        null|         5232|  11|2014-12-12|
         * +--------+---------+-------------+------------+-------------+----+----------+
         */
        dataset = dataset.withColumn("date", functions.substring(col("time"),0, 10));
        dataset =dataset.withColumn("time", functions.regexp_replace(col("time"), col("time"), functions.substring(col("time"),12, 2)));

        /**
         * 查看每列缺失值，可以看出user_geohash缺失较多，该列数据暂时用不上，故可删除
         * +-------+-------+-------------+------------+-------------+----+----+
         * |user_id|item_id|behavior_type|user_geohash|item_category|time|date|
         * +-------+-------+-------------+------------+-------------+----+----+
         * |      0|      0|            0|    15911010|            0|   0|   0|
         * +-------+-------+-------------+------------+-------------+----+----+
         */
        String[] columnsName = dataset.columns();
        Column[] columns = new Column[columnsName.length];
        for(int index = 0;index < columnsName.length; index++) {
            columns[index] = functions.count(functions.when(functions.isnull(col(columnsName[index])), columnsName[index])).as(columnsName[index]);
        }
        //dataset.select(columns).show();

        /**删除列user_geohash
         * +--------+---------+-------------+-------------+----+----------+
         * | user_id|  item_id|behavior_type|item_category|time|      date|
         * +--------+---------+-------------+-------------+----+----------+
         * |10001082|285259775|            1|         4076|  18|2014-12-08|
         * |10001082|  4368907|            1|         5503|  12|2014-12-12|
         * |10001082|  4368907|            1|         5503|  12|2014-12-12|
         * |10001082| 53616768|            1|         9762|  15|2014-12-02|
         * |10001082|151466952|            1|         5232|  11|2014-12-12|
         * +--------+---------+-------------+-------------+----+----------+
         *  |-- user_id: integer (nullable = true)
         *  |-- item_id: integer (nullable = true)
         *  |-- behavior_type: integer (nullable = true)
         *  |-- item_category: integer (nullable = true)
         *  |-- time: string (nullable = true)
         *  |-- date: string (nullable = true)
         *
         */
        dataset = dataset.drop("user_geohash");

        /**
         * 增加一列，将一天24小时划分为：'凌晨','上午','中午','下午','晚上'
         *      '凌晨'：0-5
         *      '上午'：5-10
         *      '中午'：10-13
         *      '下午'：13-18
         *      '晚上'：18-24
         *
         * +--------+---------+-------------+-------------+----+----------+----------+
         * | user_id|  item_id|behavior_type|item_category|time|      date|time_slice|
         * +--------+---------+-------------+-------------+----+----------+----------+
         * |10001082|285259775|            1|         4076|  18|2014-12-08|      晚上|
         * |10001082|  4368907|            1|         5503|  12|2014-12-12|      中午|
         * |10001082|  4368907|            1|         5503|  12|2014-12-12|      中午|
         * |10001082| 53616768|            1|         9762|  15|2014-12-02|      下午|
         * |10001082|151466952|            1|         5232|  11|2014-12-12|      中午|
         * +--------+---------+-------------+-------------+----+----------+----------+
         */

        dataset = dataset.map((MapFunction<Row, Row>) row -> {
            String timeSlice;
            int time = Integer.parseInt(row.getString(4).trim());
            if(time >= 0 && time < 5) {
                timeSlice = "凌晨";
            } else if (time >= 5 && time < 10) {
                timeSlice = "上午";
            } else if (time >= 10 && time < 13) {
                timeSlice = "中午";
            } else if (time >= 13 && time < 18) {
                timeSlice = "下午";
            } else {
                timeSlice = "晚上";
            }
            return RowFactory.create(row.getInt(0), row.getInt(1), row.getInt(2), row.getInt(3), row.getString(4), row.getString(5), timeSlice);
        }, RowEncoder.apply(new StructType(new StructField[]{
                new StructField("user_id", DataTypes.IntegerType,false, Metadata.empty()),
                new StructField("item_id", DataTypes.IntegerType,false, Metadata.empty()),
                new StructField("behavior_type", DataTypes.IntegerType,false, Metadata.empty()),
                new StructField("item_category", DataTypes.IntegerType,false, Metadata.empty()),
                new StructField("time", DataTypes.StringType,false, Metadata.empty()),
                new StructField("date", DataTypes.StringType,false, Metadata.empty()),
                new StructField("time_slice", DataTypes.StringType,false, Metadata.empty())
        })));

        //app流量分析
        appTrafficAnalysis(dataset);

        //用户行为分析
        userBehaviorAnalysis(session, dataset);

        //商品分析
        productAnalysis(dataset);
    }