Spark Streaming快速入门系列(5) | foreachRDD输出

小咪咪 2022-12-04 05:28 161阅读 0赞

**`foreachRDD：将于把数据写出到外部存储`**

*  1.需求  
    使用foreachRDD写入到MySQL中
 *  2.代码实现

import java.util.Properties
    
    import org.apache.spark.SparkConf
    import org.apache.spark.sql.SparkSession
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    
    object OutDemo1 { 
      val props = new Properties()
      props.setProperty("user","root")
      props.setProperty("password","root")
      def main(args: Array[String]): Unit = { 
        val conf = new SparkConf().setMaster("local[*]").setAppName("Transform")
        val ssc = new StreamingContext(conf, Seconds(3))
        ssc.checkpoint("ck2")
    
        ssc
          .socketTextStream("hadoop102", 9999)
          .flatMap(_.split("\\W+"))
          .map((_, 1))
            .updateStateByKey((seq: Seq[Int], opt: Option[Int])=>Some(seq.sum+opt.getOrElse(0)))
           /* .foreachRDD(rdd=>{ rdd.foreachPartition(it=>{ //连接到mysqk //写数据 //关闭mysql }) })*/
          .foreachRDD(rdd=>{ 
              //把rdd转成df
              //1.先创建sparkSession
              val spark = SparkSession.builder()
                .config(rdd.sparkContext.getConf)
                .getOrCreate()
              import spark.implicits._
              //2.转换
              val df = rdd.toDF("word", "count")
              //3.写
              df.write.mode("overwrite").jdbc("jdbc:mysql://hadoop102:3306/test","word1602",props)
            })
    
    
        ssc.start()
        ssc.awaitTermination()
    
      }
    }

*  hadoop102上启动 netcat

nc -lk 9999

![在这里插入图片描述][20200906104129505.png_pic_center]

*  查看结果  
    ![\- \[ \]][- _]

[20200906104129505.png_pic_center]: /images/20221123/3db4e8d3a39147cba430cf730f620a74.png
[- _]: /images/20221123/53f27b2aeae044f3bf50fda6823886a2.png