package com.itechthink.rdd

import org.apache.spark.{SparkConf, SparkContext}

/**
 * Scala中常用的Action算子
 *
 * reduce：			聚合计算
 * collect：		获取元素集合
 * take(n)：		获取前n个元素
 * count：			获取元素总数
 * saveAsTextFile：	保存文件
 * countByKey：		统计相同的key出现多少次
 */
object ScalaActionOperator {
	private def getSparkContext: SparkContext = {
		val conf = new SparkConf()
			// 设置应用程序名称
			.setAppName("ScalaActionOperator")
			// 设置Spark运行模式
			.setMaster("local")
		new SparkContext(conf)
	}

	def main(args: Array[String]): Unit = {
		// 1. 先创建SparkContext
		val sc = getSparkContext

		// reduce：聚合计算
		//reduceOperator(sc);

		// collect：获取元素集合
		//collectOperator(sc);

		// take(n)：获取前n个元素
		//takeOperator(sc);

		// count：获取元素总数
		//countOperator(sc);

		// saveAsTextFile：保存文件
		//saveAsTextFileOperator(sc);

		// countByKey：统计相同的key出现多少次
		//countByKeyOperator(sc);

		// 2. 关闭SparkContext
		sc.stop()
	}

	private def countByKeyOperator(sc: SparkContext): Unit = {
		val rdd = sc.parallelize(Array(("a", 1), ("b", 2), ("a", 3), ("b", 4)))
		println(rdd.countByKey())
	}

	private def saveAsTextFileOperator(sc: SparkContext): Unit = {
		val rdd = sc.parallelize(1 to 10)
		rdd.saveAsTextFile("/Users/bear/Downloads/result")
	}

	private def countOperator(sc: SparkContext): Unit = {
		val rdd = sc.parallelize(1 to 10)
		println(rdd.count())
	}

	private def takeOperator(sc: SparkContext): Unit = {
		val rdd = sc.parallelize(1 to 10)
		println(rdd.take(5).mkString(","))
	}

	private def collectOperator(sc: SparkContext): Unit = {
		val rdd = sc.parallelize(1 to 10)
		println(rdd.collect().mkString(","))
	}

	private def reduceOperator(sc: SparkContext): Unit = {
		val rdd = sc.parallelize(1 to 10)
		println(rdd.reduce(_ + _))
	}
}

package com.itechthink.rdd

import org.apache.spark.{RangePartitioner, SparkConf, SparkContext}

/**
 * Scala中常用的Transformation算子
 *
 * map：         对集合中每个元素乘以2
 * filter：      过滤出集合中的偶数
 * flatMap：     将行拆分为单词
 * groupByKey：  对数据进行分组
 * reduceByKey： 分组统计数量
 * sortByKey：   分组进行排序
 * join：        两个集合的笛卡尔积
 * distinct：    数据去重
 */
object ScalaTransformationOperator {
	private def getSparkContext: SparkContext = {
		val conf = new SparkConf()
			// 设置应用程序名称
			.setAppName("ScalaTransformationOperator")
			// 设置Spark运行模式
			.setMaster("local")
		new SparkContext(conf)
	}

	def main(args: Array[String]): Unit = {
		// 1. 先创建SparkContext
		val sc = getSparkContext

		// map：对集合中每个元素乘以2
		//mapOperator(sc)

		// filter：过滤出集合中的偶数
		//filterOperator(sc)

		// flatMap：将行拆分为单词
		//flatMapOperator(sc)

		// groupByKey：对数据进行分组
		//groupByKeyOperator(sc)

		// reduceByKey：分组统计数量
		//reduceByKeyOperator(sc)

		// sortByKey：分组进行排序
		//sortByKeyOperator(sc)

		// join：两个集合的笛卡尔积
		//joinOperator(sc)

		// distinct：数据去重
		distinctOperator(sc)

		// 2. 关闭SparkContext
		sc.stop()
	}

	private def distinctOperator(sc: SparkContext): Unit = {
		val rdd = sc.parallelize(Array(1, 2, 2, 4, 5))
		rdd.distinct().foreach(println)
	}

	private def joinOperator(sc: SparkContext): Unit = {
		val rdd1 = sc.parallelize(Array(("a", 1), ("b", 2), ("c", 3)))
		val rdd2 = sc.parallelize(Array(("a", 4), ("b", 5), ("c", 6)))
		rdd1.join(rdd2).foreach(println)
	}

	private def sortByKeyOperator(sc: SparkContext): Unit = {
		val rdd = sc.parallelize(Array(("a", 1), ("b", 2), ("c", 3), ("a", 4), ("b", 5)))

		// 通过RangePartitioner可以实现全局排序
		val pairs = rdd.sortByKey()
		val part = new RangePartitioner(2, pairs, false)
		println(part.getPartition(("a")))
		println(part.getPartition(("b")))

		rdd.sortByKey().foreach(println)
	}

	private def reduceByKeyOperator(sc: SparkContext): Unit = {
		val rdd = sc.parallelize(Array(("a", 1), ("b", 2), ("c", 3), ("a", 4), ("b", 5)))
		rdd.reduceByKey(_ + _).foreach(println)
	}

	private def groupByKeyOperator(sc: SparkContext): Unit = {
		val rdd = sc.parallelize(Array(("a", 1), ("b", 2), ("c", 3), ("a", 4), ("b", 5)))
		rdd.groupByKey().foreach(println)
	}

	private def flatMapOperator(sc: SparkContext): Unit = {
		val rdd = sc.parallelize(Array("hello world", "hello scala", "hello java"))
		rdd.flatMap(_.split(" ")).foreach(println)
	}

	private def filterOperator(sc: SparkContext): Unit = {
		val rdd = sc.parallelize(Array(1, 2, 3, 4, 5))
		rdd.filter(_ % 2 == 0).foreach(println)
	}

	private def mapOperator(sc: SparkContext): Unit = {
		val rdd = sc.parallelize(Array(1, 2, 3, 4, 5))
		rdd.map(_ * 2).foreach(println)
	}
}