/*
 * Copyright (c) Microsoft Corporation
 * <p/>
 * All rights reserved. 
 * <p/>
 * MIT License
 * <p/>
 * Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
 * documentation files (the "Software"), to deal in the Software without restriction, including without limitation
 * the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and
 * to permit persons to whom the Software is furnished to do so, subject to the following conditions:
 * <p/>
 * The above copyright notice and this permission notice shall be included in all copies or substantial portions of
 * the Software. 
 * <p/>
 * THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO
 * THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
 * TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE. 
 */

package com.microsoft.azure.hdinsight.spark.mock.jobapp

import org.apache.spark._ 

object WordCountTest {
  def main(args: Array[String]) {
    val inputFile = args(0)
    val outputFileOption = if (args.length <= 1) None else Some(args(1))
    val conf = new SparkConf().setAppName("wordCount")
    // Create a Scala Spark Context. 
    val sc = new SparkContext(conf)
    // Load our input data. 
    val input =  sc.textFile(inputFile)
    // Split up into words. 
    val words = input.flatMap(line => line.split(" "))
    // Transform into word and count. 
    val counts = words.map(word => (word, 1)).reduceByKey{case (x, y) => x + y}

    outputFileOption match {
      // Save the word count back out to a text file, causing evaluation. 
      case Some(outputFile) => {
        val dfs = org.apache.hadoop.fs. FileSystem.get(new java.net. URI(outputFile), sc.hadoopConfiguration)
        dfs.delete(new org.apache.hadoop.fs. Path(outputFile), true)

        counts.saveAsTextFile(outputFile)
      }
      // just output to stdout
      case None =>
        counts.collect().foreach { case (word, count) => println(s"$word,$count") }
    }
  }
}
