title: Spark Streaming Saving data to MySQL with foreachRDD() in Scala
body: <p>Spark Streaming Saving data to MySQL with <code>foreachRDD()</code> in Scala</p>
  
  <p>Please, can somebody give me a functional example about saving an Spark Streaming to MySQL DB using <code>foreachRDD()</code> in Scala. I have below code but it's not working. I just need a simple example, not sintaxis or theory. </p>
  
  <p>Thank you!</p>
  
  <pre><code>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 =&gt; line.split(",")  )
      val input = split.map(x =&gt; x(0))
      input.foreachRDD { rdd =&gt;
                   if (rdd.take (1).size == 1) {
                       rdd.foreachPartition { iterator =&gt;
                           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()      
    }   
  }
  </code></pre>