import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
 
object SparkWordCount {
  def FILE_NAME: String = "word_count_results_";
  
  def main(args:Array[String]) {
    if (args.length < 1) {
      println("Usage:SparkWordCount FileName");
      System.exit(1);
    }
    val conf = new SparkConf().setAppName("Spark Exercise: Spark Version Word Count Program");
    val sc = new SparkContext(conf);
    val textFile = sc.textFile(args(0));
    // 核心步骤：Step1 + Step2 + Step3
    val wordCounts = textFile.flatMap(line => line.split(" ")).map(
      word => (word, 1)).reduceByKey((a, b) => a + b) 
    
    //print the results,for debug use.
    //println("Word Count program running results:");
    //wordCounts.collect().foreach(e => {
    //val (k,v) = e
    //println(k+"="+v)
    //});
    wordCounts.saveAsTextFile(FILE_NAME+System.currentTimeMillis());
    println("Word Count program running results are successfully saved.");
  }
}

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
object AvgAgeCalculator {
  def main(args:Array[String]) {
    if (args.length < 1){
      println("Usage:AvgAgeCalculator datafile")
      System.exit(1)
    }
    val conf = new SparkConf().setAppName("Spark Exercise:Average Age Calculator")
    val sc = new SparkContext(conf)
    val dataFile = sc.textFile(args(0), 5);
    val count = dataFile.count()        // Step2
    val ageData = dataFile.map(line => line.split(" ")(1))    // Step1
    val totalAge = ageData.map(age => Integer.parseInt(
                                String.valueOf(age))).collect().reduce((a,b) => a+b)   // Step3
    println("Total Age:" + totalAge + ";Number of People:" + count )
    val avgAge : Double = totalAge.toDouble / count.toDouble      // Step4
    println("Average Age is " + avgAge)
 }
}

object PeopleInfoCalculator {
  def main(args:Array[String]) {
    if (args.length < 1){
      println("Usage:PeopleInfoCalculator datafile")
      System.exit(1)
    }
    val conf = new SparkConf().setAppName("Spark Exercise:People Info(Gender & Height) Calculator")
    val sc = new SparkContext(conf)
    val dataFile = sc.textFile(args(0), 5);
    val maleData = dataFile.filter(line => line.contains("M")).map(
          line => (line.split(" ")(1) + " " + line.split(" ")(2)))
    val femaleData = dataFile.filter(line => line.contains("F")).map(
          line => (line.split(" ")(1) + " " + line.split(" ")(2)))            // Step1
    //for debug use
    //maleData.collect().foreach { x => println(x)}
    //femaleData.collect().foreach { x => println(x)}
    val maleHeightData = maleData.map(line => line.split(" ")(1).toInt)
    val femaleHeightData = femaleData.map(line => line.split(" ")(1).toInt)   // Step2
    //for debug use
    //maleHeightData.collect().foreach { x => println(x)}
    //femaleHeightData.collect().foreach { x => println(x)}
    val lowestMale = maleHeightData.sortBy(x => x,true).first()
    val lowestFemale = femaleHeightData.sortBy(x => x,true).first()
    //for debug use
    //maleHeightData.collect().sortBy(x => x).foreach { x => println(x)}
    //femaleHeightData.collect().sortBy(x => x).foreach { x => println(x)}
    val highestMale = maleHeightData.sortBy(x => x, false).first()
    val highestFemale = femaleHeightData.sortBy(x => x, false).first()      // Step3
    println("Number of Male Peole:" + maleData.count())
    println("Number of Female Peole:" + femaleData.count())
    println("Lowest Male:" + lowestMale)
    println("Lowest Female:" + lowestFemale)
    println("Highest Male:" + highestMale)
    println("Highest Female:" + highestFemale)
  }
}

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
 
object TopKSearchKeyWords {
  def main(args:Array[String]){
    if (args.length < 2) {
      println("Usage:TopKSearchKeyWords KeyWordsFile K");
      System.exit(1)
    }
    val conf = new SparkConf().setAppName("Spark Exercise:Top K Searching Key Words")
    val sc = new SparkContext(conf)
    val srcData = sc.textFile(args(0))
    val countedData = srcData.map(line => (line.toLowerCase(),1)).reduceByKey((a,b) => a+b)     // Step1 Step2
    //for debug use
    //countedData.foreach(x => println(x))
    val sortedData = countedData.map{ case (k,v) => (v,k) }.sortByKey(false)      // Step3
    val topKData = sortedData.take(args(1).toInt).map{ case (v,k) => (k,v) }      // Step4
    topKData.foreach(println)
  }
}

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.types._
import org.apache.spark.sql.Row
import org.apache.spark.rdd.RDD

object PeopleDataStatistics2 {
  private val schemaString = "id,gender,height"
  
  def main(args: Array[String]) {
    if (args.length < 1) {
      println("Usage:PeopleDataStatistics2 filePath")
      System.exit(1)
    }
    
    val conf = new SparkConf().setAppName("Spark Exercise:People Data Statistics 2")
    val sc = new SparkContext(conf)
    val peopleDataRDD = sc.textFile(args(0))
    val sqlCtx = new SQLContext(sc)
    
    // this is used to implicitly convert an RDD to a DataFrame.
    import sqlCtx.implicits._
    
    val schemaArray = schemaString.split(",")     // 结构化信息：列名
    val schema = StructType(schemaArray.map(fieldName => StructField(fieldName, StringType, true)))
    val rowRDD: RDD[Row] = peopleDataRDD.map(_.split(" ")).map(eachRow => Row(eachRow(0), eachRow(1), eachRow(2)))
    val peopleDF = sqlCtx.createDataFrame(rowRDD, schema)
    peopleDF.registerTempTable("people")      // 将 DataFrame peopleDF 注册为临时表
    
    //1. get the male people whose height is more than 180
    val higherMale180 = sqlCtx.sql("select id,gender,height from people where height > 180 and gender='M'")
    println("Men whose height are more than 180: " + higherMale180.count())
    println("<Display #1>")
    
    //2. get the female people whose height is more than 170
    val higherFemale170 = sqlCtx.sql("select id,gender,height from people where height > 170 and gender='F'")
    println("Women whose height are more than 170: " + higherFemale170.count())
    println("<Display #2>")
    
    //3. Grouped the people by gender and count the number
    peopleDF.groupBy(peopleDF("gender")).count().show()
    println("People Count Grouped By Gender")
    println("<Display #3>")
 
    //4. 用类 RDD 转换的方式对 DataFrame 操作来统计并打印身高大于 210cm 的前 50 名男性
    peopleDF.filter(peopleDF("gender").equalTo("M")).filter(peopleDF("height") > 210).show(50)
    println("Men whose height is more than 210")
    println("<Display #4>")
 
    //5. 对所有人按身高进行排序并打印前 50 名的信息
    peopleDF.sort($"height".desc).take(50).foreach { row => println(row(0) + "," + row(1) + "," + row(2)) }
    println("Sorted the people by height in descend order,Show top 50 people")
    println("<Display #5>")
 
    //6. 统计男性的平均身高
    peopleDF.filter(peopleDF("gender").equalTo("M")).agg(Map("height" -> "avg")).show()
    println("The Average height for Men")
    println("<Display #6>")
 
    //7. 统计女性身高的最大值
    peopleDF.filter(peopleDF("gender").equalTo("F")).agg("height" -> "max").show()
    println("The Max height for Women:")
    println("<Display #7>")

    println("All the statistics actions are finished on structured People data.")
  }
}

import org.apache.spark.sql.SQLContext
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkContext, SparkConf}

//define case class for user（用户）
case class User(userID: String, gender: String, age: Int, registerDate: String,role: String, region: String)
//define case class for consuming data（交易）
case class Order(orderID: String, orderDate: String, productID: Int, price: Int, userID: String)

object UserConsumingDataStatistics {
  def main(args: Array[String]) {
    if (args.length < 1) {
      println("Usage:UserConsumingDataStatistics userDataFilePath consumingDataFilePath")
      System.exit(1)
    }
    val conf = new SparkConf().setAppName("Spark Exercise:User Consuming Data Statistics")
    //Kryo serializer is more quickly by default java serializer
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    val ctx = new SparkContext(conf)
    val sqlCtx = new SQLContext(ctx)
    import sqlCtx.implicits._

    //Convert user data RDD to a DataFrame and register it as a temp table
    val userDF = ctx.textFile(args(0)).map(_.split(" ")).map(u => User(u(0), u(1), u(2).toInt,u(3),u(4),u(5))).toDF()
    userDF.registerTempTable("user")
 
    //Convert consuming data RDD to a DataFrame and register it as a temp table
    val orderDF = ctx.textFile(args(1)).map(_.split(" ")).map(o => Order(o(0), o(1), o(2).toInt,o(3).toInt,o(4))).toDF()
    orderDF.registerTempTable("orders")
 
    //cache the DF in memory with serializer should make the program run much faster
    userDF.persist(StorageLevel.MEMORY_ONLY_SER)
    orderDF.persist(StorageLevel.MEMORY_ONLY_SER)

    //1. The number of people who have orders in the year 2015（统计在2015年有交易的用户人数）
    val count = orderDF.filter(orderDF("orderDate").contains("2015")).join(userDF, orderDF("userID").equalTo(userDF("userID"))).count()
    println("The number of people who have orders in the year 2015:" + count)
    
    //2. total orders produced in the year 2014（在2014年的总订单量）
    val countOfOrders2014 = sqlCtx.sql("SELECT * FROM orders where orderDate like '2014%'").count()
    println("total orders produced in the year 2014:" + countOfOrders2014)
 
    //3. Orders that are produced by user with ID 1 information overview（用户ID为1的用户的订单信息）
    val countOfOrdersForUser1 = sqlCtx.sql("SELECT o.orderID,o.productID,o.price,u.userID FROM orders o,user u 
                              where u.userID =1 and u.userID = o.userID").show()
    println("Orders produced by user with ID 1 showed.")
 
    //4. Calculate the max,min,avg prices for the orders that are producted by user with ID 10（用户ID为10的用户交易的最大、最小和平均价格）
    val orderStatsForUser10 = sqlCtx.sql("SELECT max(o.price) as maxPrice,min(o.price) as minPrice,avg(o.price) as avgPrice,u.userID 
                              FROM orders o,user u where u.userID = 10 and u.userID = o.userID group by u.userID")
    println("Order statistic result for user with ID 10:")
 
    orderStatsForUser10.collect().map(order => "Minimum Price=" + order.getAs("minPrice")
                              + ";Maximum Price=" + order.getAs("maxPrice")
                              + ";Average Price=" + order.getAs("avgPrice")
                              ).foreach(result => println(result))
 }
}