发表更新大数据1 小时读完 (大约7082个字)0次访问

Spark —— 淘宝双 11 数据分析

问题定义

实验内容

使用 Spark 对数据进行处理，并分析双十一的用户交易等信息，用 Scala 语言进行程序编写，最后将处理的数据结果使用 Echarts 进行可视化。

实验环境以及使用的相关应用：

  • Spark 2.4.3
  • Scala 2.12
  • Tomcat 9.0.20
  • 可视化工具：ECharts
  • Java 包：fastjson
  • 系统环境：macOS Mojave 10.14.5

数据集

本案列主要分析淘宝双十一的数据，数据集是淘宝 2015 年双十一前 6 个月（包含双十一）的交易数据，数据集 user_log.csv，是记录了用户的行为的日志文件。

日志 user_log.csv 的字段定义如下：

序号 字段 定义
0 user_id 买家 id
1 item_id 商品 id
2 cat_id 商品类别 id
3 merchant_id 卖家 id
4 brand_id 品牌 id
5 month 交易时间：月
6 day 交易事件：日
7 action 行为，取值范围 {0,1,2,3}，0 表示点击，1 表示加入购物车，2 表示购买，3 表示关注商品
8 age_range 买家年龄分段：1 表示年龄 <18；2 表示年龄在 [18,24]；3 表示年龄在 [25,29]；4 表示年龄在 [30,34]；5 表示年龄在 [35,39]；6 表示年龄在 [40,49]；7 和 8 表示年龄 >=50；0 和 NULL 则表示未知
9 gender 性别：0 表示女性，1 表示男性，2 和 NULL 表示未知
10 province 收货地址省份

前五条记录样例：

1
2
3
4
5
user_id,item_id,cat_id,merchant_id,brand_id,month,day,action,age_range,gender,province
328862,323294,833,2882,2661,08,29,0,0,1,内蒙古
328862,844400,1271,2882,2661,08,29,0,1,1,山西
328862,575153,1271,2882,2661,08,29,0,2,1,山西
328862,996875,1271,2882,2661,08,29,0,1,1,内蒙古

SparkRDD 实验过程及其结果

实验一：查看日志前 10 个交易日志的商品品牌

分析：使用 take() 函数，取去掉首行后的前 10 行数据，按条件为 brand_id 的项进行筛选，并逐个输出

  1. 使用 take() 函数取出去掉首行后的前 10 行数据
  2. 进行 map 操作，将数据处理成 ($商品品牌) 样式
  3. 输出 map 后的结果

核心代码如下：

1
2
3
4
5
6
7
8
9
10
def transformationOps(sc: SparkContext): Unit = {
  // 查看日志前10个交易日志的商品品牌
  val raw_data = sc.textFile("./data/user_log.csv")
  val header = raw_data.first()
  val lines = raw_data.filter(row => row != header)

  lines.take(10).flatMap(line => line.split(" "))
    .map(line => (line.split(",")(4)))
    .foreach(println)
}

结果如图：

实验二：查询前 20 个交易日志中购买商品时的时间和商品的种类

分析：使用 take() 函数，取去掉首行后的前 20 行数据，按条件为 month, day, cat_id 的项进行筛选，并逐个输出。（这里将 month 和 day 合并为一个字符串）

  1. 使用 take() 函数取出去掉首行后的前 20 行数据
  2. 进行 map 操作，将数据处理成 ($日期, $商品种类) 样式
  3. 输出 map 后的结果

核心代码如下：

1
2
3
4
5
6
7
8
9
10
11
def transformationOps(sc: SparkContext): Unit = {
  // 查询前20个交易日志中购买商品时的时间和商品的种类
  val raw_data = sc.textFile("./data/user_log.csv")
  val header = raw_data.first()
  val lines = raw_data.filter(row => row != header)
  val data = lines.take(20)
    .flatMap(line => line.split(" "))
    .map(line => (line.split(",")(5) +line.split(",")(6), line.split(",")(2)))
    
  data.foreach(println)
}

结果如图：

实验三：查询双十一那天有多少人购买了商品

分析：

  1. 首先进行 map 操作，将数据处理成 ($日期, $用户ID, $用户行为) 样式
  2. 然后使用 filter() 函数过滤出 11 月 11 日所有用户行为为购买的条目，具体为日期 month == 11, day == 11，行为 action == 2
  3. 考虑到单个用户可能购买多个商品，所以需要去重，使用 distinct() 函数去重
  4. 最后使用 count() 函数计数并输出

核心代码如下：

1
2
3
4
5
6
7
8
9
10
11
def transformationOps(sc: SparkContext): Unit = {
  // 查询双十一那天有多少人购买了商品
  val raw_data = sc.textFile("./data/user_log.csv")
  val Users = raw_data.flatMap(line => line.split(" "))
    .map(line => (line.split(",")(5) + line.split(",")(6), line.split(",")(0), line.split(",")(7)))
  val output = Users.filter(line => line._3.contains("2"))
    .filter(line => line._1.contains("1111"))
    .distinct()
  
  println("双十一有 " + output.count() + " 人购买了商品。")
}

结果如图：

实验四：取给定时间和给定品牌，求当天购买的此品牌商品的数量

分析：

  1. 首先进行 map 操作，将数据处理成 ($品牌ID, $日期, $用户行为) 样式
  2. 然后使用 filter() 函数过滤出 11 月 11 日品牌 2661 中用户行为为购买的条目，具体为日期 month == 11, day == 11，品牌 IDbrand_id == 2661，行为 action == 2
  3. 使用 count() 函数计数并输出

核心代码如下：

1
2
3
4
5
6
7
8
9
10
11
def transformationOps(sc: SparkContext): Unit = {
  // 取给定时间和给定品牌，求当天购买的此品牌商品的数量
  val raw_data = sc.textFile("./data/user_log.csv")
  val data = raw_data.flatMap(line => line.split(" "))
    .map(line => (line.split(",")(4), line.split(",")(5) + line.split(",")(6), line.split(",")(7)))
  val output = data.filter(line => line._1.contains("2661"))
    .filter(line => line._2.contains("1111"))
    .filter(line => line._3.contains("2"))

  println("11月11日，品牌2661的商品售出的数量为：" + output.count())
}

结果如图：

实验五：查询有多少用户在双十一点击了该店

分析：

  1. 首先进行 map 操作，将数据处理成 ($用户ID, $卖家ID, $日期, $用户行为) 样式
  2. 然后使用 filter() 函数过滤出 11 月 11 日店铺 2882 中用户行为为点击的用户，具体为日期 month == 11, day == 11，卖家 IDmerchant_id == 2882，行为 action == 0
  3. 考虑到每个用户可能多次点击同一店铺，所以需要去重，使用 distinct() 函数。

核心代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def transformationOps(sc: SparkContext): Unit = {
  // 查询有多少用户在双十一点击了该店
  val raw_data = sc.textFile("./data/user_log.csv")
  val data = raw_data.flatMap(line => line.split(" "))
    .map(line => (line.split(",")(0),   // 用户ID
                  line.split(",")(3),   // 卖家ID
                  line.split(",")(5) + line.split(",")(6),  // 日期
                  line.split(",")(7)))  // 行为
  val output = data.filter(line => line._3.contains("1111"))
    .filter(line => line._2.contains("2882"))
    .filter(line => line._4.contains("0")).distinct()

  println("一共有" + output.count() + "个用户在双十一点击了店铺2882")
}

结果如图：

实验六：查询双十一那天女性购买商品的数量

分析：统计同时满足日期 month == 11, day == 11，行为 action == 2，性别 gender == 0 的个数。

  1. 首先进行 map 操作，将数据处理成 (($用户ID, $日期, $用户行为), 1) 样式
  2. 然后使用 filter() 函数过滤出 11 月 11 日用户行为为购买的用户，具体为日期 month == 11, day == 11，行为 action == 2
  3. 然后使用 reduceByKey() 统计所有用户的购买量，然后使用 filter() 筛选出大于 5 次的用户，并逐个输出

核心代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
def transformationOps(sc: SparkContext): Unit = {
  // 查询双十一那天女性购买商品的数量
  val raw_data = sc.textFile("./data/user_log.csv")
  val data = raw_data.flatMap(line => line.split(" "))
    .map(line => (line.split(",")(5) + line.split(",")(6),  // 日期
                  line.split(",")(7),   // 行为
                  line.split(",")(9)))  // 性别
  val res = data.filter(line => line._1.contains("1111"))
    .filter(line => line._2.contains("2"))
    .filter(line => line._3.contains("0"))

  println("双十一那天女性购买商品的数量为：" + res.count())
}

结果如图：

实验七：查询双十一那天男性购买商品的数量

分析：

  1. 首先进行 map 操作，将数据处理成 ($日期, $用户行为, $性别) 样式
  2. 然后使用 filter() 函数过滤出 11 月 11 日用户行为为购买的男性用户，具体为日期 month == 11, day == 11，行为 action == 2，性别 gender == 1
  3. 然后使用 count() 函数统计所有过滤后的数据量，并输出

核心代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
def transformationOps(sc: SparkContext): Unit = {
  // 查询双十一那天男性购买商品的数量
  val raw_data = sc.textFile("./data/user_log.csv")
  val data = raw_data.flatMap(line => line.split(" "))
    .map(line => (line.split(",")(5) + line.split(",")(6),  // 日期
                  line.split(",")(7),   // 行为
                  line.split(",")(9)))  // 性别
  val res = data.filter(line => line._1.contains("1111"))
    .filter(line => line._2.contains("2"))
    .filter(line => line._3.contains("1"))

  println("双十一那天男性购买商品的数量为：" + res.count())
}

结果如图：

实验八：查询某一天在该网站购买商品超过 5 次的用户 ID

分析：

  1. 首先进行 map 操作，将数据处理成 (($用户ID, $日期, $用户行为), 1) 样式
  2. 然后使用 filter() 函数过滤出 11 月 11 日用户行为为购买的用户，具体为日期 month == 11, day == 11，行为 action == 2
  3. 然后使用 reduceByKey() 统计所有用户的购买量，然后使用 filter() 筛选出大于 5 次的用户，并逐个输出

核心代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
def transformationOps(sc: SparkContext): Unit = {
  // 查询某一天在该网站购买商品超过5次的用户ID
  val raw_data = sc.textFile("./data/user_log.csv")
  val data = raw_data.flatMap(line => line.split(" "))
    .map(line => ((line.split(",")(0), //用户ID
      line.split(",")(5) + line.split(",")(6), // 日期
      line.split(",")(7)), 1)) // 行为
  val lines = data.filter(line => line._1._2.contains("1111")).filter(line => line._1._3.contains("2"))
  val Users = lines.reduceByKey((v1, v2) => v1 + v2).filter(t => t._2 > 5)

  Users.foreach(userID => println("用户 " + userID._1._1 + " 在 " + userID._1._2 + " 购买了 " + userID._2 + " 件商品"))
  println("11月11日购买超过5次的用户一共有：" + Users.count() + "个")
}

结果如图：

使用 ECharts 进行可视化

ECharts，一个使用 JavaScript 实现的开源可视化库，可以流畅的运行在 PC 和移动设备上，兼容当前绝大部分浏览器（IE8/9/10/11，Chrome，Firefox，Safari 等），底层依赖轻量级的矢量图形库 ZRender，提供直观，交互丰富，可高度个性化定制的数据可视化图表。

首先使用 SparkRDD 处理我们需要的数据，使用阿里巴巴的 Java 库 fastjson，将 SparkRDD 输出的数据转换成 JSON 格式的文档。然后使用 Tomcat 搭建 JSP 服务，用 jQuery 将 *.json 文件内的数据导入到 ECharts，最后呈现数据。

可视化实验一：双十一所有买家消费行为比例

数据处理

分析：

  1. 首先进行 map 操作，将数据处理成 ($日期, $用户行为) 样式
  2. 然后使用 filter() 函数确定时间为 11 月 11 日，具体为日期 month == 11, day == 11
  3. 使用 map() 函数将数据处理为 ($用户行为, 1)
  4. 使用 reduceByKey() 计算出每个用户行为对应的条目数
  5. 将结果转换成 JSON 格式文件，并输出处理结果

数据处理核心代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def transformationOps(sc: SparkContext): Unit = {
  // 双十一所有买家消费行为比例
  val raw_data = sc.textFile("./data/user_log.csv")
  val data = raw_data.flatMap(line => line.split("\n"))
    .map(line => ((line.split(",")(5) + line.split(",")(6), // 日期
      line.split(",")(7)), 1)) // 行为
    .filter(line => line._1._1.equals("1111"))
    .map(t => (t._1._2, t._2))
  val Output = data.reduceByKey((v1, v2) => v1 + v2).collect()
  
  toJSON_Pie(Output, sc)	// 输出成JSON格式
  
  Output.foreach(println)
}

转换成 JSON 文件

编写 toJSON_Pie() 函数，将 SparkRDD 处理的结果转换成 JSON 格式的文件，用于生成饼状图

  1. 使用阿里巴巴 Java 库 fastjson，新建一个 JSONObject 对象
  2. 新建一个数组准备存放绘制 EChart 饼状图要求的 {"name":"","value":""} 对
  3. 使用循环，按行读取，放入 SparkRDD 处理的结果
  4. 导出，保存

函数代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def toJSON_Pie(data: Array[(String, Int)], sc: SparkContext): Unit = {
  val JSONObject = new JSONObject()
  val m = data.length
  val list: util.List[Any] = new util.ArrayList[Any]()
  for (i <- 0 until m) {
    val pair = new util.HashMap[String, Any]()
    pair.put("name", data(i)._1.toString)
    pair.put("value", data(i)._2.toString)
    list.add(pair)
  }
  JSONObject.put("data", list)
  val output = sc.parallelize(List(JSONObject))
  output.repartition(1).saveAsTextFile("../Visualization/web/data/v_1")
}

导出后的文件为 Spark 处理的文件，名称为 part-00000，需要手动将其加上 json 后缀为 part-00000.json

由于原始数据中代表用户行为的数据为 0~3，数字对可视化呈现不直观，所以手动更改了 json 文件的对应内容

part-00000.json 文件内容为：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
{
  "data": [
    {
      "name": "点击",
      "value": "9188104"
    },
    {
      "name": "加入购物车",
      "value": "14725"
    },
    {
      "name": "购买",
      "value": "1223354"
    },
    {
      "name": "关注",
      "value": "156450"
    }
  ]
}

可视化呈现

使用 Tomcat 搭建本地 Java Web Server，编写 jsp 网页，将 *.json 文件导入，使用 ECharts 呈现

根据 ECharts 官方文档：异步数据加载和更新，使用 jQuery 工具将 json 文件导入

核心代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
var myChart = echarts.init(document.getElementById('main'), 'shine');

myChart.setOption({
    title: {
        text: '双十一所有买家消费行为比例',
        x: 'center'
    },
    tooltip: {
        trigger: 'item',
        formatter: "{a} <br/> {b} : {c} ({d}%)"
    },
    legend: {
        orient: 'vertical',
        right: 50,
        top: 150,
        // bottom: 20,
        data: ['点击', '加入购物车', '购买', '关注']
    },
    label: {
        formatter: '{b}: {c}'
    },
    series: [
        {
            name: '行为',
            type: 'pie',
            radius: '60%',
            center: ['40%', '50%'],
            data: [],
            selectedMode: 'single',
            itemStyle: {
                emphasis: {
                    shadowBlur: 10,
                    shadowOffsetX: 0,
                    shadowColor: 'rgba(0, 0, 0, 0.5)'
                }
            }
        }
    ]
});

$.get('./data/v_1/part-00000.json').done(function (data) {
    myChart.setOption({
        series: [{
            data: data.data
        }]
    });
});

可视化结果如图所示：

可视化实验二：双十一当天销量前十的商品类别

数据处理

分析：

  1. 首先进行 map 操作，将数据处理成 ($日期, $商品类别, $用户行为) 样式
  2. 使用 filter() 函数确定时间为 11 月 11 日，用户行为为购买的数据，具体为日期 month == 11, day == 11，用户行为 action == 2
  3. 使用 map() 函数将数据处理为 ($商品类别, 1)
  4. 使用 reduceByKey() 计算出每个商品类别对应的条目数
  5. 使用 sortBy() 按条目数量降序排序
  6. 使用 take() 取前 10 个数据，然后将结果转换成 JSON 格式文件，并输出处理结果

数据处理核心代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def transformationOps(sc: SparkContext): Unit = {
  // 双十一当天销量前十的商品类别
  val raw_data = sc.textFile("./data/user_log.csv")
  val data = raw_data.flatMap(line => line.split("\n"))
    .map(line => ((line.split(",")(5) + line.split(",")(6), // 日期
      line.split(",")(2), // 商品类别
      line.split(",")(7)), 1))
    .filter(_._1._1.equals("1111"))
    .filter(_._1._3.equals("2"))
    .map(t => (t._1._2, t._2))
  val Output = data.reduceByKey((v1, v2) => v1 + v2)
    .sortBy(_._2, ascending = false, numPartitions = 1).collect()

  toJSON_Bar(Output.take(10), sc)

  Output.take(10).foreach(println)
}

转换成 JSON 文件

编写 toJSON_Bar() 函数，将 SparkRDD 处理的结果转换成 JSON 格式的文件，用于生成柱状图

  1. 使用阿里巴巴 Java 库 fastjson，新建一个 JSONObject 对象
  2. 新建两个数组准备存放绘制 EChart 柱状图要求的数据
  3. 使用循环，按行读取，放入 SparkRDD 处理的结果
  4. 导出，保存

函数代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def toJSON_Bar(data: Array[(String, Int)], sc: SparkContext): Unit = {
  val JSONObject = new JSONObject()
  val m = data.length
  val key: Array[Any] = new Array[Any](m)
  val value: Array[Any] = new Array[Any](m)
  for (i <- 0 until m) {
    key(i) = data(i)._1.toString
    value(i) = data(i)._2.toString
  }
  JSONObject.put("key", key)
  JSONObject.put("value", value)
  val output = sc.parallelize(List(JSONObject))
  output.repartition(1).saveAsTextFile("../Visualization/web/data/v_2")
}

导出后的文件为 Spark 处理的文件，名称为 part-00000，需要手动将其加上 json 后缀为 part-00000.json

part-00000.json 文件内容为：

1
2
3
4
{
  "value":["48687","41951","29531","28883","28530","28097","27372","26264","25851","23519"],
  "key":["656","1208","602","662","737","1142","389","177","1213","1438"]
}

其中，key 表示商品种类，value 表示对应商品种类的销量

可视化呈现

使用 Tomcat 搭建本地 Java Web Server，编写 jsp 网页，将 *.json 文件导入，使用 ECharts 呈现

根据 ECharts 官方文档：异步数据加载和更新，使用 jQuery 工具将 json 文件导入

核心代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
// 基于准备好的dom，初始化echarts实例
var myChart = echarts.init(document.getElementById('main'), 'shine');

myChart.setOption({
    title: {
        text: '双十一当天销量前10的商品类别',
        // x: 'center'
    },
    legend: {
        data: ["销量"],
        top: '3'
    },
    tooltip: {
        trigger: 'axis',
        axisPointer: {            // 坐标轴指示器，坐标轴触发有效
            type: 'shadow'        // 默认为直线，可选为：'line' | 'shadow'
        }
    },
    grid: {
        left: '3%',
        right: '4%',
        bottom: '3%',
        containLabel: true
    },
    xAxis: [
        {
            type: 'category',
            data: [],
            axisTick: {
                alignWithLabel: true
            }
        }
    ],
    yAxis: [
        {
            type: 'value'
        }
    ],
    series: [
        {
            name: '销量',
            type: 'bar',
            barWidth: '60%',
            data: [],
            label: {
                normal: {
                    show: true,
                    position: 'top'
                }
            }
        }
    ]
});

$.get('./data/v_2/part-00000.json').done(function (data) {
    myChart.setOption({
        xAxis: [{
            data: data.key
        }],
        series: [{
            data: data.value
        }]
    });
});

可视化结果如图所示：

可视化实验三：双十一男女买家各个年龄段交易对比

数据处理

分析：

  1. 首先进行 map 操作，将数据处理成 (($日期, $用户行为, $年龄段, $性别), 1) 样式
  2. 然后使用 filter() 函数确定时间为 11 月 11 日，用户行为为购买，具体为日期 month == 11, day == 11，用户行为 action == 2
  3. 使用 map() 函数将数据处理为 (($年龄段, $性别), 1)
  4. 使用 reduceByKey() 计算出每个年龄段每个性别对应的条目数
  5. 为了方便后续生成对应的 JSON 格式数据，使用两个 sortBy() 函数将数据按年龄段从小到大，性别 “女，男，未知” 顺序排序
  6. 将结果转换成 JSON 格式文件，并输出处理结果

数据处理核心代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def transformationOps(sc: SparkContext): Unit = {
  // 双十一男女买家各个年龄段交易对比
  val raw_data = sc.textFile("./data/user_log.csv")
  val data = raw_data.flatMap(line => line.split("\n"))
    .map(line => ((line.split(",")(5) + line.split(",")(6), // 日期
      line.split(",")(7), // 行为
      line.split(",")(8), // 年龄段
      line.split(",")(9)), 1)) // 性别
    .filter(line => line._1._1.equals("1111"))
    .filter(line => line._1._2.equals("2"))
    .map(t => ((t._1._3, t._1._4), t._2))
  val Output = data.reduceByKey((v1, v2) => v1 + v2)
    .sortBy(_._1._2).sortBy(_._1._1).collect()

  toJSON_MultiBar(Output, sc)

  Output.foreach(println)
}

转换成 JSON 文件

编写 toJSON_MultiBar() 函数，将 SparkRDD 处理的结果转换成 JSON 格式的文件，用于生成柱状图

  1. 使用阿里巴巴 Java 库 fastjson，新建一个 JSONObject 对象
  2. 新建四个数组分别存放：年龄段；女性、男性、其他在各个年龄段的交易量
  3. 使用循环，按行读取，按照不同性别筛选，将 SparkRDD 处理的结果放入对应的数组
  4. 导出，保存

函数代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def toJSON_MultiBar(data: Array[((String, String), Int)], sc: SparkContext): Unit = {
  val JSONObject = new JSONObject()
  val m = data.length
  val age: Array[Any] = new Array[Any](8)
  val woman: Array[Any] = new Array[Any](8)
  val man: Array[Any] = new Array[Any](8)
  val other: Array[Any] = new Array[Any](8)

  for (j <- 0 to 7) {
    age(j) = j.toString   // 7个对应的年龄段
  }
  JSONObject.put("age", age)
  
  var x1 = 0
  var x2 = 0
  var x3 = 0

  for (i <- 0 until m) {
    if (data(i)._1._2 == "0") {       
      woman(x1) = data(i)._2.toString   // 将性别女对应的条目写入 woman 数组
      x1 = x1 + 1
    }
    else if (data(i)._1._2 == "1") {  
      man(x2) = data(i)._2.toString     // 将性别男对应的条目写入 man 数组
      x2 += 1
    }
    else {                              
      other(x3) = data(i)._2.toString   // 将性别未知对应的条目写入 other 数组
      x3 += 1
    }
  }
  
  JSONObject.put("woman", woman)
  JSONObject.put("man", man)
  JSONObject.put("other", other)

  val output = sc.parallelize(List(JSONObject))
  output.repartition(1).saveAsTextFile("./Visualization/web/data/v_3")
}

导出后的文件为 Spark 处理的文件，名称为 part-00000，需要手动将其加上 json 后缀为 part-00000.json

part-00000.json 文件内容为：

1
2
3
4
5
6
{
  "other":["51126","50628","50989","51169","50916","50706","51183","51029"],
  "woman":["50590","51337","50854","50978","51259","51244","51517","50872"],
  "man":["50855","50732","51195","51076","50798","50555","50855","50891"],
  "age":["0","1","2","3","4","5","6","7"]
}

可视化呈现

使用 Tomcat 搭建本地 Java Web Server，编写 jsp 网页，将 *.json 文件导入，使用 ECharts 呈现

根据 ECharts 官方文档：异步数据加载和更新，使用 jQuery 工具将 json 文件导入

核心代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
// 基于准备好的dom，初始化echarts实例
var myChart = echarts.init(document.getElementById('main'), 'shine');

$.get('./data/v_3/part-00000.json').done(function (data) {
    myChart.setOption({
        title: {
            text: '双十一男女买家各个年龄段交易对比',
            x: 'center'
        },
        legend: {
            data: ["女", "男", "未知"],
            right: '20',
            top: '30'
        },
        tooltip: {
            trigger: 'axis',
            axisPointer: {            // 坐标轴指示器，坐标轴触发有效
                type: 'shadow'        // 默认为直线，可选为：'line' | 'shadow'
            }
        },
        grid: {
            left: '3%',
            right: '4%',
            bottom: '3%',
            containLabel: true
        },
        xAxis: [
            {
                type: 'category',
                data: data.age,
                axisTick: {
                    alignWithLabel: true
                }
            }
        ],
        yAxis: [
            {
                type: 'value'
            }
        ],
        series: [
            {
                type: 'bar',
                name: '女',
                data: data.woman
            },
            {
                type: 'bar',
                name: '男',
                data: data.man
            },
            {
                type: 'bar',
                name: '未知',
                data: data.other
            }
        ]
    });
});

可视化结果如图所示：

可视化实验四：双十一男女买家交易对比

数据处理

分析：

  1. 首先进行 map 操作，将数据处理成 (($日期, $用户行为, $性别), 1) 样式
  2. 然后使用 filter() 函数确定时间为 11 月 11 日，用户行为为购买，具体为日期 month == 11, day == 11，用户行为 action == 2
  3. 使用 map() 函数将数据处理为 ($性别, 1)
  4. 使用 reduceByKey() 计算出每个性别对应的条目数
  5. 将结果转换成 JSON 格式文件，并输出处理结果

数据处理核心代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
def transformationOps(sc: SparkContext): Unit = {
  // 双十一男女买家交易对比
  val raw_data = sc.textFile("./data/user_log.csv")
  val data = raw_data.flatMap(line => line.split("\n"))
    .map(line => ((line.split(",")(5) + line.split(",")(6), // 日期
      line.split(",")(7), // 行为
      line.split(",")(9)), 1)) // 性别
    .filter(line => line._1._1.equals("1111"))
    .filter(line => line._1._2.equals("2"))
    .map(t => (t._1._3, t._2))
  val Output = data.reduceByKey((v1, v2) => v1 + v2).collect()

  toJSON_Pie(Output, sc)

  Output.foreach(println)
}

转换成 JSON 文件

编写 toJSON_Pie() 函数，将 SparkRDD 处理的结果转换成 JSON 格式的文件，用于生成饼状图

  1. 使用阿里巴巴 Java 库 fastjson，新建一个 JSONObject 对象
  2. 新建一个数组准备存放绘制 EChart 饼状图要求的 {"name":"","value":""} 对
  3. 使用循环，按行读取，放入 SparkRDD 处理的结果
  4. 导出，保存

函数代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def toJSON_Pie(data: Array[(String, Int)], sc: SparkContext): Unit = {
  val JSONObject = new JSONObject()
  val m = data.length
  val list: util.List[Any] = new util.ArrayList[Any]()
  for (i <- 0 until m) {
    val pair = new util.HashMap[String, Any]()
    pair.put("name", data(i)._1.toString)
    pair.put("value", data(i)._2.toString)
    list.add(pair)
  }
  JSONObject.put("data", list)
  val output = sc.parallelize(List(JSONObject))
  output.repartition(1).saveAsTextFile("./Visualization/web/data/v_4")
}

导出后的文件为 Spark 处理的文件，名称为 part-00000，需要手动将其加上 json 后缀为 part-00000.json

由于原始数据中代表用户行为的数据为 0~3，数字对可视化呈现不直观，所以手动更改了 json 文件的对应内容

part-00000.json 文件内容为：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
{
  "data": [
    {
      "name": "女",
      "value": "408651"
    },
    {
      "name": "男",
      "value": "406957"
    },
    {
      "name": "未知",
      "value": "407746"
    }
  ]
}

可视化呈现

使用 Tomcat 搭建本地 Java Web Server，编写 jsp 网页，将 *.json 文件导入，使用 ECharts 呈现

根据 ECharts 官方文档：异步数据加载和更新，使用 jQuery 工具将 json 文件导入

核心代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
// 基于准备好的dom，初始化echarts实例
var myChart = echarts.init(document.getElementById('main'), 'shine');

// $.getJSON('./data/v_1/part-00000.json').done(function (data) {
myChart.setOption({
    title: {
        text: '双十一男女买家交易对比',
        x: 'center'
    },
    tooltip: {
        trigger: 'item',
        formatter: "{a} <br/> {b}: {c} ({d}%)"
    },
    label: {
        formatter: "{b}: {c}"
    },
    legend: {
        orient: 'vertical',
        right: 50,
        top: 150,
        // bottom: 20,
        data: ['女', '男', '未知']
    },
    series: [
        {
            name: '性别',
            type: 'pie',
            radius: '60%',
            center: ['40%', '50%'],
            data: [],
            selectedMode: 'single',
            itemStyle: {
                emphasis: {
                    shadowBlur: 10,
                    shadowOffsetX: 0,
                    shadowColor: 'rgba(0, 0, 0, 0.5)'
                }
            }
        }
    ]
});

$.get('./data/v_4/part-00000.json').done(function (data) {
    myChart.setOption({
        series: [{
            data: data.data
        }]
    });
});

可视化结果如图所示：

可视化实验五：各个省份的总成交量对比

数据处理

分析：

  1. 首先进行 map 操作，将数据处理成 (($省份, $用户行为), 1) 样式
  2. 然后使用 filter() 函数确定用户行为为购买，具体为用户行为 action == 2
  3. 使用 map() 函数将数据处理为 ($省份, 1)
  4. 使用 reduceByKey() 计算出每个省份对应的条目数
  5. 将结果转换成 JSON 格式文件，并输出处理结果

数据处理核心代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
def transformationOps(sc: SparkContext): Unit = {
  // 各个省份的总成交量对比
  val raw_data = sc.textFile("./data/user_log.csv")
  val data = raw_data.flatMap(line => line.split("\n"))
    .map(line => ((line.split(",")(10), line.split(",")(7)), 1))
    .filter(line => line._1._2.contains("2"))
    .map(t => (t._1._1, t._2))
  val Output = data.reduceByKey((v1, v2) => v1 + v2).collect()

  toJSON_Map(Output, sc)

  Output.foreach(println)
}

转换成 JSON 文件

编写 toJSON_Pie() 函数，将 SparkRDD 处理的结果转换成 JSON 格式的文件，用于生成饼状图

  1. 使用阿里巴巴 Java 库 fastjson，新建一个 JSONObject 对象
  2. 新建一个数组准备存放绘制 EChart 地图要求的 {"name":"","value":""} 对
  3. 使用循环，按行读取，放入 SparkRDD 处理的结果
  4. 导出，保存

函数代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def toJSON_Map(data: Array[(String, Int)], sc: SparkContext): Unit = {
  val JSONObject = new JSONObject()
  val m = data.length
  val list: util.List[Any] = new util.ArrayList[Any]()
  for (i <- 0 until m) {
    val pair = new util.HashMap[String, Any]()
    pair.put("name", data(i)._1.toString)
    pair.put("value", data(i)._2.toString)
    list.add(pair)
  }
  JSONObject.put("data", list)
  val output = sc.parallelize(List(JSONObject))
  output.repartition(1).saveAsTextFile("./Visualization/web/data/v_5")
}

导出后的文件为 Spark 处理的文件，名称为 part-00000，需要手动将其加上 json 后缀为 part-00000.json

part-00000.json 文件内容为：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
{
  "data":[
    {"name":"天津","value":"97134"},
    {"name":"吉林","value":"96992"},
    {"name":"四川","value":"96743"},
    {"name":"内蒙古","value":"96619"},
    {"name":"上海市","value":"96893"},
    {"name":"浙江","value":"97162"},
    {"name":"青海","value":"96991"},
    {"name":"河北","value":"96674"},
    {"name":"云南","value":"96377"},
    {"name":"香港","value":"96587"},
    {"name":"宁夏","value":"96494"},
    {"name":"甘肃","value":"96822"},
    {"name":"安徽","value":"96488"},
    {"name":"江西","value":"96679"},
    {"name":"江苏","value":"96802"},
    {"name":"湖南","value":"97125"},
    {"name":"西藏","value":"97116"},
    {"name":"福建","value":"96625"},
    {"name":"北京","value":"96911"},
    {"name":"陕西","value":"96517"},
    {"name":"山西","value":"96861"},
    {"name":"广西","value":"96675"},
    {"name":"澳门","value":"96710"},
    {"name":"湖北","value":"96943"},
    {"name":"山东","value":"97123"},
    {"name":"广东","value":"97338"},
    {"name":"台湾","value":"96543"},
    {"name":"辽宁","value":"96714"},
    {"name":"新疆","value":"96960"},
    {"name":"黑龙江","value":"97169"},
    {"name":"重庆","value":"96813"},
    {"name":"海南","value":"96722"},
    {"name":"河南","value":"96980"},
    {"name":"贵州","value":"96842"}
  ]
}

可视化呈现

使用 Tomcat 搭建本地 Java Web Server，编写 jsp 网页，将 *.json 文件导入，使用 ECharts 呈现

根据 ECharts 官方文档：异步数据加载和更新，使用 jQuery 工具将 json 文件导入

核心代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
// 基于准备好的dom，初始化echarts实例
var myChart = echarts.init(document.getElementById('main'), 'shine');
myChart.showLoading();

$.getJSON('./data/china.json', function (geoJson) {

    myChart.hideLoading();

    echarts.registerMap('CN', geoJson);

    myChart.setOption({
        title: {
            text: '各个省份的总成交量对比',
            x: 'center'
        },
        tooltip: {
            trigger: 'item',
            formatter: "{b} <br/> {c} (件)"
        },
        toolbox: {
            show: true,
            orient: 'vertical',
            left: 'right',
            top: 'center',
            feature: {
                dataView: {readOnly: false},
                restore: {},
                saveAsImage: {}
            }
        },
        visualMap: {
            min: 96000,
            max: 98000,
            text: ['高', '低'],
            realtime: false,
            calculable: true,
            inRange: {
                color: ['lightskyblue', 'yellow', 'orangered']
            }
        },
        series: [
            {
                name: '各个省份的总成交量对比',
                type: 'map',
                map: 'CN', // 自定义扩展图表类型
                itemStyle: {
                    normal: {label: {show: true}},
                    emphasis: {label: {show: true}}
                },
                data: []
            }
        ]
    })
});


$.get('./data/v_5/part-00000.json').done(function (data) {
    myChart.setOption({
        series: [{
            data: data.data
        }]
    });
});

可视化结果如图所示：

实验过程中发现的问题

在本次实验过程中发现了不少问题，有些通过在网络搜索解决了，但有些没有找到很好的解决方法，在此提出。

  • 在编写可视化部分的代码时，使用 SparkRDD 生成的 JSON 格式的文件没有后缀，这时 jQuery 无法读取到文件的内容，必须手动更改 part-00000 文件名为 part-00000.json
  • 没有找到合适的方法修改 SparkRDD 里的数据，当绘制可视化实验一：双十一所有买家消费行为比例和可视化实验四：双十一男女买家交易对比时，由于原始数据中代表用户行为的数据为 0 ~ 3，代表用户性别的数据为 0 ~ 1，数字对可视化呈现不直观，所以手动需要更改 json 文件的对应内容。同样在绘制可视化实验五：各个省份的总成交量对比时，数据里有些省份的名字和 china.js 中的省份名称不一样，比如北京和北京市，这样可视化的时候数据会不显示，也需要手动更改 JSON 文件内容。

Spark —— 淘宝双 11 数据分析

https://morooi.com/2019/spark-taobao/

作者

SJ Zhou

发布于

2019-06-07

更新于

2021-01-06

许可协议

#

评论

分类

最新文章

×