val temp = transactions.flatMap(_.map((_, 1)))
  .reduceByKey(_ + _)
  .filter { case (_, count) => count >= minCount }
  // 根据支持度计数由大到小排列
  .sortBy(-_._2)
  .cache()
// 频繁1项集，用作索引映射元素值
val l1 = temp.map(_._1).collect()
// 事务集中的元素值映射索引
val item2Rank = l1.zipWithIndex.toMap

val l1BC = sc.broadcast(l1)
val tmp = transactions
  // 事务压缩，过滤掉不包含频繁1项集的事务项
  .filter(_.exists(l1BC.value.contains))
  // 过滤掉不包含频繁1项集的元素
  .map(_.filter(l1BC.value.contains))
  // 统计重复事务项
  .map(x => (x.toSet, 1)).reduceByKey(_ + _)
// 事务计数表
var cntMap = tmp.map(_._2).collect()

// 构造布尔矩阵
var boolMatrix = tmp.map(_._1)
  .map { transaction =>
    val item2Rank = item2RankBC.value
    val boolRow = new Array[Boolean](l1.length)
    transaction.foreach { x =>
      boolRow(item2Rank(x)) = true
    }

    boolRow
  }.collect()

val kItem = freqItems.collect()
val candidates = collection.mutable.ListBuffer.empty[Set[Int]]
kItem.indices.foreach { i =>
    Range(i + 1, kItem.length).foreach { j =>
        if ((kItem(i) & kItem(j)).size == k - 2) {
            candidates.append(kItem(i) | kItem(j))
        }
    }
}
// 剪枝，频繁项集的子集也是频繁项集
sc.parallelize(candidates.distinct.filter(x => kItem.exists(_.subsetOf(x))))

val boolMatrixBC = sc.broadcast(boolMatrix)
val cntMapBC = sc.broadcast(cntMap)
// 统计候选项对应事务的布尔向量
val tmp = candidates.map { item =>
    val boolMatrix = boolMatrixBC.value
    val boolVector = item.map(x => boolMatrix.map(_ (x)))
        // 布尔向量间做按位与
        .reduce(_.zip(_).map(x => x._1 & x._2))

    (item, boolVector)
}.cache()
// 支持度计数
val candidatesWithCnt = tmp.map { case (item, vector) =>
    val cntMap = cntMapBC.value
    val count = vector.zip(cntMap).map { case (flag, count) =>
        if (flag) count else 0
    }.sum

(item, count)
}
// 事务的使用表
val usedMap = tmp.map(_._2)
     // 布尔向量间做按位或
     .reduce(_.zip(_).map(x => x._1 | x._2))
// 事务压缩
val newBoolMatrix = boolMatrix.zip(usedMap).filter(_._2).map(_._1)
val newCntMap = cntMap.zip(usedMap).filter(_._2).map(_._1)

val loop = new Breaks
loop.breakable {
    while (!lkWithCnt.isEmpty) {
        // k阶候选集
        val candidates = genCandidates(sc, lkWithCnt.map(_._1), k, minCount)
          .persist(StorageLevel.MEMORY_AND_DISK)
        if (candidates.isEmpty) {
          loop.break()
        }
        // 事务压缩，计算支持度计数
        val (candidatesWithCnt, newBoolMatrix, newCntMap) = calCount(sc, candidates, boolMatrix, cntMap)
        boolMatrix = newBoolMatrix
        cntMap = newCntMap
        // 支持度过滤
        lkWithCnt = candidatesWithCnt.filter { case (_, count) => count >= minCount }
          .persist(StorageLevel.MEMORY_AND_DISK)
        freqItemsWithCnt = freqItemsWithCnt.union(lkWithCnt)
        candidates.unpersist()
        k += 1
    }
}