目錄

SparkStreaming實戰案例六 自定義輸出-foreachRDD

需求

注意:

代碼實現

---

SparkStreaming實戰案例六 自定義輸出-foreachRDD

需求

對上述案例的結果數據輸出到控制臺外的其他組件,如MySQL/HDFS

注意:

foreachRDD函數屬于將DStream中結果數據RDD輸出的操作，類似transform函數，針對每批次RDD數據操作，但無返回值

DStream.print方法源碼底層調用的也是foreachRDD：

?

?

代碼實現

package cn.itcast.streamingimport java.sql.{Connection, DriverManager, PreparedStatement, Timestamp}import org.apache.commons.lang3.time.FastDateFormat
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}/*** 使用SparkStreaming接收Socket數據,node01:9999* 對上述案例的結果數據輸出到控制臺外的其他組件,如MySQL/HDFS*/
object SparkStreamingDemo06_Output {def main(args: Array[String]): Unit = {//1.創建環境val conf: SparkConf = new SparkConf().setAppName(this.getClass.getSimpleName.stripSuffix("$")).setMaster("local[*]")//設置數據輸出文件系統的算法版本為2//https://blog.csdn.net/u013332124/article/details/92001346.set("spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version", "2")val sc: SparkContext = new SparkContext(conf)sc.setLogLevel("WARN")val ssc: StreamingContext = new StreamingContext(sc, Seconds(5))ssc.checkpoint("./ckp")//2.接收socket數據val linesDS: ReceiverInputDStream[String] = ssc.socketTextStream("node1",9999)//3.做WordCountval wordAndCountDS: DStream[(String, Int)] = linesDS.flatMap(_.split(" ")).map((_, 1))//windowDuration:窗口長度:就算最近多久的數據,必須都是微批間隔的整數倍//slideDuration :滑動間隔:就是每隔多久計算一次,,必須都是微批間隔的整數倍//每隔10s(slideDuration :滑動間隔)計算最近20s(windowDuration:窗口長度)的熱搜排行榜!.reduceByKeyAndWindow((v1:Int, v2:Int)=>v1+v2, Seconds(20),Seconds(10))//排序取TopN//注意:DStream沒有直接排序的方法!所以應該調用DStream底層的RDD的排序方法!//transform(函數),該函數會作用到DStream底層的RDD上!val resultDS: DStream[(String, Int)] = wordAndCountDS.transform(rdd => {val sortedRDD: RDD[(String, Int)] = rdd.sortBy(_._2, false)val top3: Array[(String, Int)] = sortedRDD.take(3) //取出當前RDD中排好序的前3個熱搜詞!println("======top3--start======")top3.foreach(println)println("======top3--end======")sortedRDD})//4.輸出resultDS.print()resultDS.foreachRDD((rdd,time)=>{val df: FastDateFormat = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss")val batchTime: String = df.format(time.milliseconds)println("-------自定義的輸出-------")println(s"batchTime:${batchTime}")println("-------自定義的輸出-------")if(!rdd.isEmpty()){//-1.輸出到控制臺rdd.foreach(println)//-2.輸出到HDFSrdd.coalesce(1).saveAsTextFile(s"hdfs://node1:8020/wordcount/output-${time.milliseconds}")//-3.輸出到MySQL/*CREATE TABLE `t_hotwords` (`time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,`word` varchar(255) NOT NULL,`count` int(11) DEFAULT NULL,PRIMARY KEY (`time`,`word`)) ENGINE=InnoDB DEFAULT CHARSET=utf8;*/rdd.foreachPartition(iter=>{val conn: Connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/bigdata?characterEncoding=UTF-8","root","root")val sql:String = "REPLACE INTO `t_hotwords` (`time`, `word`, `count`) VALUES (?, ?, ?);"val ps: PreparedStatement = conn.prepareStatement(sql)//獲取預編譯語句對象iter.foreach(t=>{val word: String = t._1val count: Int = t._2ps.setTimestamp(1,new Timestamp(time.milliseconds) )ps.setString(2,word)ps.setInt(3,count)ps.addBatch()})ps.executeBatch()ps.close()conn.close()})}})//5.啟動并等待程序停止ssc.start()ssc.awaitTermination()ssc.stop(stopSparkContext = true, stopGracefully = true)}
}

?

總結

以上是生活随笔為你收集整理的2021年大数据Spark（四十一）：SparkStreaming实战案例六 自定义输出 foreachRDD的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。