Fork me on GitHub

Spark应用案例-关于购物篮的设计

发表于 | 分类于 | 阅读次数

介绍

购物篮的定义



适用场景




　
### 相关概念 ###



　

### 步骤 ###



　　

### 编程实现 ###



　
　
### 步骤 ###



　
　　

## 程序实现 ##

### 程序 ###

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
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197

import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

/**
  * 使用SparkCore实现购物篮分析
  */
object FindAssociationRulesSparkCore {
  /**
    * 先从缓存中获取数据，如果不存在，直接重新获取
    *
    * @param items
    * @param size
    * @param cache
    * @return
    */
  def findItemSetsByCache(items: List[(String, Int)], size: Int, cache: mutable.Map[Int, List[List[(String, Int)]]]): List[List[(String, Int)]] = {
    cache.get(size).orElse {
      // 获取值
      val result = findItemSets(items, size, cache)

      // 更新缓存
      cache += size -> result

      // 返回值
      Some(result)
    }.get
  }


  /**
    * 构建项集基于items商品列表，项集中的商品数量是size指定
    *
    * @param items 商品列表：eg: [A, B, C]
    * @param size  最终项集包含商品的数量
    * @return
    */
  def findItemSets(items: List[(String, Int)], size: Int, cache: mutable.Map[Int, List[List[(String, Int)]]]): List[List[(String, Int)]] = {
    if (size == 1) {
      // items中的每个商品都是一个项集
      items.map(item => item :: Nil)
    } else {
      // 当size不是1的时候
      // 1. 获取项集大小为size-1的项集列表
      val tmpItemSets = findItemSetsByCache(items, size - 1, cache)
      // 2. 给tmpItemSets中添加一个新的不重复的项 ==> 数据的转换
      val itemSets = tmpItemSets.flatMap(itemSets => {
        // 给itemSets项集添加一个新的商品ID，要求不重复
        val newItemSets = items
          // 将包含的商品过滤掉&要求下标必须大于以及存在
          .filter(item => !itemSets.contains(item) && itemSets.forall(_._2 < item._2))
          // 将商品添加到项集中，产生一个新的项集
          // 为了使用distinct做去重操作，进行一个排序操作
          .map(item => (item :: itemSets))

        // 返回值
        newItemSets
      })

      // 返回项集的值
      itemSets
    }
  }

  def main(args: Array[String]): Unit = {
    // 1. 创建SparkContext
    val conf = new SparkConf()
      .setAppName("find-association-rules")
      .setMaster("local[*]")
    val sc = SparkContext.getOrCreate(conf)

    // ===========================================
    // 测试数据存储的路径
    val path = "data/transactions/10"
    val savePath = "data/transactions/result"
    // 最小支持度
    val minSupport = 2
    // 最小置信度
    val minConfidence = 0.4

    // 创建rdd读取原始的交易数据，
    // 假设交易数据是按行存储的，每行是一条交易，每条交易数据包含的商品ID使用","分割
    val rdd = sc.textFile(path, 20)

    // 1. 计算频繁项集
    // 1.1 获取每条交易存在的项集
    val itemSetsRDD: RDD[String] = rdd.flatMap(transaction => {
      // 1) 获取当前交易所包含的商品ID
      val items = transaction
        .split(",") // 分割
        .filter(!_.isEmpty) // 过滤
        .sorted //排序
        .toList // 转换为list
        .zipWithIndex // 将数据和下标合并，下标从0开始

      // 2) 构建辅助对象
      val itemSize = items.size
      val cache = mutable.Map[Int, List[List[(String, Int)]]]()

      // 3) 根据获取的商品ID的信息产生项集
      // allItemSets集合中最后数据量是:2^itemSize - 1
      val allItemSets: List[List[String]] = (1 to itemSize).map(size => {
        // 产生项集中项的数量是size的项集
        findItemSets(items, size, cache)
      }).foldLeft(List[List[String]]())((v1, v2) => {
        v2.map(_.map(_._1)) ::: v1
      })

      // 4) 返回结果
      allItemSets.map(_.mkString(","))
    })

    // 1.2 获取频繁项集
    val supportedItemSetsRDD = itemSetsRDD
      // 数据转换
      .map(items => (items, 1))
      // 聚合求支持度
      .reduceByKey(_ + _)
      // 过滤产生频繁项集
      .filter(_._2 >= minSupport)

    // 2. 计算关联规则
    // 2.1 对每个频繁项集获取子项集
    val subSupportedItemSetsRDD = supportedItemSetsRDD.flatMap(tuple => {
      val itemSets = tuple._1.split(",").toList.zipWithIndex // 频繁项集
      val frequency = tuple._2 // 该频繁项集的支持度

      // 2) 构建辅助对象
      val itemSize = itemSets.size
      val cache = mutable.Map[Int, List[List[(String, Int)]]]()

      // 3) 获取子项集
      val allSubItemSets: List[List[String]] = (1 to itemSize).map(size => {
        // 产生项集中项的数量是size的项集
        findItemSets(itemSets, size, cache)
      }).foldLeft(List[List[String]]())((v1, v2) => {
        v2.map(_.map(_._1)) ::: v1
      })

      // 4) 转换数据并输出
      val items = itemSets.map(_._1)
      allSubItemSets.map(subItemSets => {
        // (A,B,frequency) ==> 表示A出现的时候B也出现的次数是frequency次
        // 当subItemSets就是itemSets的时候，返回的二元组的第二个元素的(元组)第一个元素是空的列表
        (subItemSets.mkString(","), ((items.toBuffer -- subItemSets).toList.mkString(","), frequency))
      })
    })

    // 2.2 计算置信度
    val assocRulesRDD = subSupportedItemSetsRDD
      .groupByKey() // 数据聚合
      .flatMap(tuple => {
      // 计算执行度: (A, B, k) => A存在的时候B也存储的几率是k
      // A就是tuple的第一个元素
      // 获取左件
      val lhs = tuple._1.split(",").mkString("<", ",", ">")

      // 获取左件在所有的交易中出现的总的次数 tuple._2中第一个元素为空的数据就是总的次数
      val frequency = tuple._2
        // 只要第一个元素为空的值，表示from本身
        .filter(_._1.isEmpty)
        // 需要的是第二个元素
        .map(_._2).toList match {
        case head :: Nil => head
        case _ => {
          throw new IllegalArgumentException("异常")
        }
      }

      // 计算右件出现次数占左件次数的百分比, 并返回最终结果
      tuple._2
        // 要求第一个数据非空
        .filter(!_._1.isEmpty)
        // 数据转换，获取置信度
        .map {
        case (rhs, support) => {
          // 计算置信度
          (lhs, rhs.split(",").mkString("<", ",", ">"), 1.0 * support / frequency)
        }
      }
    })

    // 2.3 过滤置信度太低的数据
    val resultRDD = assocRulesRDD.filter(_._3 >= minConfidence)

    // 3. RDD数据保存
    //resultRDD.collect()
    FileSystem.get(sc.hadoopConfiguration).delete(new Path(savePath), true)
    //resultRDD.repartition(1).saveAsTextFile(savePath)

    // ===========================================
    sc.stop()
  }
}


### 注意点（本地的完全运行） ###

　　不需要开启服务，也不需要上传文件，讲文件保存在本地的方式


优化程序

　　1.优化的是相集的个数

　　2.使用广播变量

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
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220

import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

/**
  * 使用SparkCore实现购物篮分析
  */
object FindAssociationRulesSparkCore {
  /**
    * 先从缓存中获取数据，如果不存在，直接重新获取
    *
    * @param items
    * @param size
    * @param cache
    * @return
    */
  def findItemSetsByCache(items: List[(String, Int)], size: Int, cache: mutable.Map[Int, List[List[(String, Int)]]]): List[List[(String, Int)]] = {
    cache.get(size).orElse {
      // 获取值
      val result = findItemSets(items, size, cache)

      // 更新缓存
      cache += size -> result

      // 返回值
      Some(result)
    }.get
  }


  /**
    * 构建项集基于items商品列表，项集中的商品数量是size指定
    *
    * @param items 商品列表：eg: [A, B, C]
    * @param size  最终项集包含商品的数量
    * @return
    */
  def findItemSets(items: List[(String, Int)], size: Int, cache: mutable.Map[Int, List[List[(String, Int)]]]): List[List[(String, Int)]] = {
    if (size == 1) {
      // items中的每个商品都是一个项集
      items.map(item => item :: Nil)
    } else {
      // 当size不是1的时候
      // 1. 获取项集大小为size-1的项集列表
      val tmpItemSets = findItemSetsByCache(items, size - 1, cache)
      // 2. 给tmpItemSets中添加一个新的不重复的项 ==> 数据的转换
      val itemSets = tmpItemSets.flatMap(itemSets => {
        // 给itemSets项集添加一个新的商品ID，要求不重复
        val newItemSets = items
          // 将包含的商品过滤掉&要求下标必须大于以及存在
          .filter(item => !itemSets.contains(item) && itemSets.forall(_._2 < item._2))
          // 将商品添加到项集中，产生一个新的项集
          // 为了使用distinct做去重操作，进行一个排序操作
          .map(item => (item :: itemSets))

        // 返回值
        newItemSets
      })

      // 返回项集的值
      itemSets
    }
  }

  def main(args: Array[String]): Unit = {
    val n = 10000
    // 1. 创建SparkContext
    val conf = new SparkConf()
      .setAppName(s"find-association-rules-${n}")
      .setMaster("local[*]")
    //      .set("spark.eventLog.enabled", "true")
    //      .set("spark.eventLog.dir","hdfs://hadoop-senior01:8020/spark-history")
    //      .set("spark.executor.memory","3g")
    val sc = SparkContext.getOrCreate(conf)

    // ===========================================
    // 测试数据存储的路径
    val path = s"data/transactions/${n}"
    val savePath = s"result2/${n}"
    // 最小支持度
    val minSupport = 2
    // 最小置信度
    val minConfidence = 0.1

    // 创建rdd读取原始的交易数据，
    // 假设交易数据是按行存储的，每行是一条交易，每条交易数据包含的商品ID使用","分割
    val rdd = sc.textFile(path, 20)

    // 过滤无效数据：对于在整个交易集合中出现比较少的商品过滤掉，先进行需要过滤的商品的RDD数据
    val minGoodCount = 3 // 要求商品在整个交易集中至少出现3次
    val needFilterGoodsRDD = rdd
      .flatMap(transaction => transaction
        .split(",")
        .filter(!_.isEmpty)
        .map(good => (good, 1))
      )
      .reduceByKey(_ + _)
      .filter(_._2 < minGoodCount)
      .map(_._1)
    // 使用广播变量将数据广播输出
    val needFilterGoods: Broadcast[List[String]] = sc.broadcast(needFilterGoodsRDD.collect().toList)

    // 1. 计算频繁项集
    // 1.1 获取每条交易存在的项集
    val itemSetsRDD: RDD[String] = rdd.flatMap(transaction => {
      // 1) 获取当前交易所包含的商品ID
      val goods: Array[String] = transaction
        .split(",") // 分割
        .filter(!_.isEmpty) // 过滤


      // 将需要过滤的数据过滤掉
      val items = (goods.toBuffer -- needFilterGoods.value)
        .sorted //排序
        .toList // 转换为list
        .zipWithIndex // 将数据和下标合并，下标从0开始

      // 2) 构建辅助对象
      // 最大的项集只允许存在5个项的，5怎么来？根据业务规则&根据运行之后的情况
      val itemSize = Math.min(items.size, 5)
      val cache = mutable.Map[Int, List[List[(String, Int)]]]()

      // 3) 根据获取的商品ID的信息产生项集
      // allItemSets集合中最后数据量是:2^itemSize - 1
      val allItemSets: List[List[String]] = (1 to itemSize).map(size => {
        // 产生项集中项的数量是size的项集
        findItemSets(items, size, cache)
      }).foldLeft(List[List[String]]())((v1, v2) => {
        v2.map(_.map(_._1)) ::: v1
      })

      // 4) 返回结果
      allItemSets.map(_.mkString(","))
    })

    // 1.2 获取频繁项集
    val supportedItemSetsRDD = itemSetsRDD
      // 数据转换
      .map(items => (items, 1))
      // 聚合求支持度
      .reduceByKey(_ + _)
      // 过滤产生频繁项集
      .filter(_._2 >= minSupport)

    // 2. 计算关联规则
    // 2.1 对每个频繁项集获取子项集
    val subSupportedItemSetsRDD = supportedItemSetsRDD.flatMap(tuple => {
      val itemSets = tuple._1.split(",").toList.zipWithIndex // 频繁项集
      val frequency = tuple._2 // 该频繁项集的支持度

      // 2) 构建辅助对象
      val itemSize = itemSets.size
      val cache = mutable.Map[Int, List[List[(String, Int)]]]()

      // 3) 获取子项集
      val allSubItemSets: List[List[String]] = (1 to itemSize).map(size => {
        // 产生项集中项的数量是size的项集
        findItemSets(itemSets, size, cache)
      }).foldLeft(List[List[String]]())((v1, v2) => {
        v2.map(_.map(_._1)) ::: v1
      })

      // 4) 转换数据并输出
      val items = itemSets.map(_._1)
      allSubItemSets.map(subItemSets => {
        // (A,B,frequency) ==> 表示A出现的时候B也出现的次数是frequency次
        // 当subItemSets就是itemSets的时候，返回的二元组的第二个元素的(元组)第一个元素是空的列表
        (subItemSets.mkString(","), ((items.toBuffer -- subItemSets).toList.mkString(","), frequency))
      })
    })

    // 2.2 计算置信度
    val assocRulesRDD = subSupportedItemSetsRDD
      .groupByKey() // 数据聚合
      .flatMap(tuple => {
      // 计算执行度: (A, B, k) => A存在的时候B也存储的几率是k
      // A就是tuple的第一个元素
      // 获取左件
      val lhs = tuple._1.split(",").mkString("<", ",", ">")

      // 获取左件在所有的交易中出现的总的次数 tuple._2中第一个元素为空的数据就是总的次数
      val frequency = tuple._2
        // 只要第一个元素为空的值，表示from本身
        .filter(_._1.isEmpty)
        // 需要的是第二个元素
        .map(_._2).toList match {
        case head :: Nil => head
        case _ => {
          throw new IllegalArgumentException("异常")
        }
      }

      // 计算右件出现次数占左件次数的百分比, 并返回最终结果
      tuple._2
        // 要求第一个数据非空
        .filter(!_._1.isEmpty)
        // 数据转换，获取置信度
        .map {
        case (rhs, support) => {
          // 计算置信度
          (lhs, rhs.split(",").mkString("<", ",", ">"), 1.0 * support / frequency)
        }
      }
    })

    // 2.3 过滤置信度太低的数据
    val resultRDD = assocRulesRDD.filter(_._3 >= minConfidence)

    // 3. RDD数据保存
    FileSystem.get(sc.hadoopConfiguration).delete(new Path(savePath), true)
    resultRDD.repartition(1).saveAsTextFile(savePath)

    // ===========================================
    sc.stop()
  }
}
注：此博客仅供博主自身复习用
-------------本文结束感谢您的阅读-------------
坚持技术分享，您的支持将鼓励我继续创作！