Spark Streaming Saving data to MySQL with foreachRDD() in Scala

Spark Streaming Saving data to MySQL with foreachRDD() in Scala

Please, can somebody give me a functional example about saving an Spark Streaming to MySQL DB using foreachRDD() in Scala. I have below code but it's not working. I just need a simple example, not sintaxis or theory.

Thank you!

package examples
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark._
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{Seconds, StreamingContext}
import StreamingContext._
import org.apache.hadoop.io.Text
import org.apache.hadoop.io.LongWritable
import org.apache.hadoop.mapred.SequenceFileOutputFormat
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.hive.HiveContext
import java.util.Properties
import org.apache.spark.sql.SaveMode  
object StreamingToMysql {
  def main(args: Array[String]) {
    val sparkConf = new SparkConf().setAppName("NetworkWordCount").setMaster("local[*]")
    val sc = new SparkContext(sparkConf) 
    val sqlContext = new SQLContext(sc)
    val hiveCtx= new HiveContext(sc)
    import hiveCtx.implicits._
    val ssc = new StreamingContext(sc, Seconds(1))
    val lines = ssc.socketTextStream("localhost", 9999)
    ssc.checkpoint("hdfs://localhost:54310/user/hduser/Streaming/logs")    
    val rdd = sc.parallelize(List(1))
    val df = rdd.toDF()
    val split = lines.map(line => line.split(",")  )
    val input = split.map(x => x(0))
    input.foreachRDD { rdd =>
                 if (rdd.take (1).size == 1) {
                     rdd.foreachPartition { iterator =>
                         iterator.foreach {
                          val connectionProperties = new Properties()
                          connectionProperties.put("user", "root")
                          connectionProperties.put("password", "admin123")
                          iterator.write.mode("append")
                        .jdbc("jdbc:mysql://192.168.100.8:3306/hadoopguide", "topics", connectionProperties) 
                         }
                 }
             }    
    }
    val connectionProperties = new Properties()
    connectionProperties.put("user", "root")
    connectionProperties.put("password", "admin123")
    df.write.mode("append")
  .jdbc("jdbc:mysql://192.168.100.8:3306/hadoopguide", "topics", connectionProperties)    
    println("Done")
    ssc.start()
    ssc.awaitTermination()      
  }   
}