spark学习

import org.apache.spark.{SparkConf, SparkContext}
//SparkConf  Spark应用程序的配置。用于将各种Spark参数设置为键值对。
//Spark功能的主入口点。SparkContext表示到Spark集群的连接，可用于在该集群上创建RDDs、累加器和广播变量
val sparkConf = new SparkConf().setAppName("Association Rule Mining Demo")
//new SparkConf()创建了一个从系统属性和类路径加载默认值的SparkConf对象  
//setAppName()为你的应用设置一个名字
val sc = new SparkContext(sparkConf)
val transactions = sc.textFile(transactionsFileName, 2).cache()
//textFile(path: String, minPartitions: Int = defaultMinPartitions)
//从HDFS、本地文件系统(在所有节点上都可用)或hadoop支持的文件系统URI中读取文本文件，并以字符串的RDD形式返回。
//minPartitions:生成的RDD的最小分区数
//cache():使用默认的存储级别(MEMORY_ONLY)来支持这个RDD。
val transactionSize = transactions.count();
//count():返回RDD中的元素数量。

def toList(transaction: String): List[String] = {
 val list = transaction.trim().split(",").toList
 list
}
//trim():从指定字符串列的两端修剪指定字符。没带参数就是去掉空格
//split():将提供的字符序列围绕','匹配项拆分,返回Array[String]
//toList():将Array[String]转换为List[Strings]
def findSortedCombinations[T](elements: List[T])(implicit B: Ordering[T]): List[List[T]] = {
 val result = elements.sorted(B).toSet[T].subsets.map(_.toList).toList
 result
}
//sorted(B):根据顺序对这个数组进行排序。是稳定排序
//toSet[T]():转换为Set
//subsets():遍历该集合所有子集的迭代器。
def removeOneItem(list: List[String], i: Int): List[String] = {
    if ((list == null) || list.isEmpty) {
        return list
    }
    if ((i < 0) || (i > (list.size - 1))) {
        return list
    }
    val cloned = list.take(i) ++ list.drop(i + 1)
    cloned
}
//take(i):选择前i个元素
//++：连接两个集合
//drpo(i+1):选择除前i+1个元素之外的所有元素。
//所以每次去掉了1个元素，如果初始数组传入0，就去掉第一个元素

val itemsets = transactions.map(toList).map(findSortedCombinations(_))
.flatMap(x => x).filter(_.size > 0).map(x => (x, 1L)).cache()
//map():通过对RDD的所有元素应用一个函数来返回一个新的RDD
//flatMap(x => x):
//filter(_.size > 0):过滤，过滤掉size=0的,也就是说去掉空集
//
val minSup = 0.4
val combined = itemsets.reduceByKey(_ + _)
.map(x => (x._1, (x._2, x._2.toDouble / transactionSize.toDouble)))
.filter(_._2._2 >= minSup).cache
//reduceByKey(_ + _):Key值相同的Value值相加
//(key,(value1,value2))
//value2>=minSup的过滤一下




val subitemsets = combined.flatMap(itemset => {
    val list = itemset._1
    val frequency = itemset._2._1
    val support = itemset._2._2
    var result = List((list, (List(""), (frequency, support))))
    if (list.size == 1) {
        result
    } 
    else {
        for (i <- 0 until list.size) {
            val listX = removeOneItem(list, i)
            val listY = list.diff(listX)
            result ++= List((listX, (listY, (frequency, support))))
        }
        result
    }
}).cache
//diff(listX):A-A∩B
//所以listY就是那个被去除掉的


val rules = subitemsets.groupByKey()
//groupByKey():和reduceByKey()好像差不多
val assocRules = rules.map(in => {
    val listX = in._1
    val listYLists = in._2.toList
    val countX = listYLists.filter(_._1(0) == "")(0)
    val newListYLists = listYLists.diff(List(countX))
    if (newListYLists.isEmpty) {
        val result = List((List(""), List(""), 0.0D, 0.0D))
        result
    } else {
        val result = newListYLists.map(t2 => (listX, t2._1, t2._2._1.toDouble/countX._2._1.toDouble,                             t2._2._2))
        result
    }
})

//最小置信度
val minConf = 0.7
//必须要大于置信度
val finalResult = assocRules.flatMap(x => x).filter(_._3 >= minConf)
//保存txt文件到输出路径
finalResult.saveAsTextFile(outputPath)

System.exit(0)