package ru.atconsulting.study

import java.io.PrintWriter

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.FileSystem
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.scalatest.{BeforeAndAfterAll, FunSuite}

import scala.collection.mutable.MutableList

/**
  * Created by glushchik on 16.02.2017.
  */
class TestBase extends FunSuite with BeforeAndAfterAll {
  private var sparkConf: SparkConf = _
  private var sc: SparkContext = _

  override protected def beforeAll(): Unit = {
    sparkConf = new SparkConf()
      .setAppName("spark-test-app")
      .setMaster("local")
    sc = new SparkContext(sparkConf)
    val conf = new Configuration()
    val fs = FileSystem.get(conf)
  }

  test("spark-test") {
    val file = sc.textFile("hosts.txt")
    val com = file.filter(s => s.contains(".com"))
    val result = countLetters(com)
    //    result.saveAsTextFile("output") - не работает
    // предполагаю из-за того, что у меня не подходящая файловая система

    val writer = new PrintWriter("output.txt")
    // так получаются иероглифы
//    result.collect().toList.foreach(s => {
//      writer.write(s+", ")
//    })
    //Так работает, но получается без разделителей
    // Как записать в файл с разделителями(, или \n)?
    result.collect().toList.foreach(writer.write)
    writer.close()
  }

  def countLetters(rdd: RDD[String]): RDD[String] = {
    val temp = rdd.collect().toList
    var result: MutableList[String] = MutableList()
    for (letter <- 97 to 122) {
      val letterCount = temp.count(s => s.contains(letter.toChar))
      result += letter.toChar + ":" + letterCount
    }
    sc.makeRDD(result)
  }

  override def afterAll(): Unit = {
    sc.stop()
  }

}