SparkStreaming实现实时WordCount程序的两种方法并将数据写入Mysql中：使用工具netcat

素颜马尾好姑娘i 2021-11-01 06:36 196阅读 0赞

# 首先需要了解的几个类 #

## StreamingContext ##

如何读取数据

## DStream ##

处理数据函数
    DStream里面存储着很多RDD

## PairDStreamFunctions ##

当处理的数据类型是二元组的时候，

DStream自动隐式转换为PairDStreamFunctions

## RDD ##

输出函数，将结果保存到外部系统
    def foreachFunc: (RDD[T], Time) => Unit = {
      (rdd: RDD[T], time: Time) => {
     		….
    }}

# NC（netcat）工具安装 #

## 介绍 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gxMDI1MzcyNjQ1_size_16_color_FFFFFF_t_70]

## 安装 ##

![在这里插入图片描述][20190810192908185.png]

## 测试 ##

监听：nc -v 主机名 9999
    发送：nc -lk 9999

# 实时词频统计 #

事先将nc服务打开，并测试可以通讯

## 代码实现方法一：使用DStream函数 ##

import java.sql.{Connection, DriverManager, PreparedStatement}
    import java.text.SimpleDateFormat
    import java.util.Date
    import org.apache.spark._
    import org.apache.spark.streaming._
    import org.apache.spark.streaming.dstream.ReceiverInputDStream
    
    object Wordount {
      def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setMaster("local[3]").setAppName("NetworkWordCount")
    
        val ssc = new StreamingContext(conf, Seconds(1))
        ssc.sparkContext.setLogLevel("WARN")
        val lines: ReceiverInputDStream[String] = ssc.socketTextStream("主机名", 9999)
    
        val words = lines.flatMap(_.split(" "))
    
        val pairs = words.map(word => (word, 1))
        val wordCounts = pairs.reduceByKey(_ + _）
       
        wordCounts.foreachRDD(rdd=>{
    
          if(!rdd.isEmpty()){
            println("------------------------------")
            rdd.coalesce(1).foreachPartition(
              item=>{
                Class.forName("com.mysql.jdbc.Driver")
                val url = "jdbc:mysql://主机名/test"
                val userName = "root"
                val password = "123456"
                var conn: Connection = null
                
                try {
                  //获取数据库连接的实例
                  conn = DriverManager.getConnection(url, userName, password)
                  val pst: PreparedStatement = conn.prepareStatement("INSERT INTO tb_wordcount(word,count) VALUES(?,?)")
                  //插入数据
                  item.foreach {
                    case (k,v) => {
                      println(s"word=$k,count=$v")
                      pst.setString(1,k)
                      pst.setInt(2,v)
                      //每天数据执行一次
                      pst.executeUpdate()
                    }
                  }
                } catch {
                  case e:Exception =>e.printStackTrace()
                } finally {
                  if(conn !=null) conn.close()
                }
              }
            )
          }
        })
        ssc.start()
        ssc.awaitTermination()
      }
    }

## 代码实现方法二：使用DStream的transform函数转换使用RDD中的高阶函数 ##

import org.apache.spark._
    import org.apache.spark.streaming._
    import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
    
    object WordCount {
      def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setMaster("local[3]").setAppName("NetworkWordCount")
    
        val ssc = new StreamingContext(conf, Seconds(1))
        ssc.sparkContext.setLogLevel("WARN")
        val lines: ReceiverInputDStream[String] = ssc.socketTextStream("bigdata-hpsk01.huadian.com", 9999)
    
    
        //尽量使用RDD的函数，使用DStream函数可能会涉及到检查点无法恢复
        val wordCountStreaming: DStream[(String, Int)] = lines.transform(rdd=>{
          val wordCount =rdd.flatMap(_.split("\\s+"))
            .filter(_.trim.length>0)
            .map((_,1))
            .reduceByKey(_ + _)
          wordCount
        })
    
        wordCountStreaming.print()
        ssc.start()
        ssc.awaitTermination()
      }
    }

# **尽量使用RDD的函数，使用DStream函数可能会涉及到检查点无法恢复** #

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gxMDI1MzcyNjQ1_size_16_color_FFFFFF_t_70]: /images/20211101/6bc6994802cf4082b1ddbadc7307b81c.png
[20190810192908185.png]: /images/20211101/81554f4fb8c846a7b52cf671b01ce9c3.png