package by.iba
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import java.sql.DriverManager
import java.sql.Connection
import java.sql.SQLException;

object Main {
  def main(args: Array[String]) {
    if(args.length<3){
      println("Not enough args")
      System.exit(1)
    }
    val driver = "com.mysql.jdbc.Driver"
    val url = "jdbc:mysql://172.20.140.245/db_stat"
    val username = "db_stat"
    val password = "pass4db!"
    var connection:Connection = null
    val conf = new SparkConf().setAppName("WordCounter").setMaster("yarn-client")
    val sc = new SparkContext(conf)
    val text_file = sc.textFile(args(0))
    val text_names = sc.textFile(args(2))
    val names = text_names.flatMap(line => line.split(",") ).collect()
    val lines = text_file.flatMap( line => line.split(" ") )
    val counts = lines.filter(line => names.contains(line)).map(word => (word, 1)).reduceByKey(_ + _)
    val summary = counts.collect()
    counts.saveAsTextFile(args(1))
        try {
      // make the connection
      Class.forName(driver)
      connection = DriverManager.getConnection(url, username, password)

      // create the statement, and run the select query
      val sql = "INSERT INTO TEST(test_name,count)VALUES(?,?);"
      val statement = connection.prepareStatement(sql)
      def ins(a:(String, Int)){
          statement.setString(1,a._1 )
          statement.setInt(2,a._2 )
          val resultSet = statement.executeUpdate()
      }
      summary.foreach { x=>ins(x) }
            } catch {
      case ex =>     println("SQLException: " + ex.getMessage())

           }
    connection.close()
  }

}