Spark Streaming算子开发代码讲解

package StreamingDemo

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * 实时黑名单过滤
 */
object TransformDemo {
 def main(args: Array[String]): Unit = {
  //设置日志级别
  Logger.getLogger("org").setLevel(Level.WARN)
  val conf = new SparkConf()
   .setAppName(this.getClass.getSimpleName)
   .setMaster("local[2]")
  val ssc = new StreamingContext(conf, Seconds(2))

  //创建一个黑名单的RDD
  val blackRDD =
   ssc.sparkContext.parallelize(Array(("zs", true), ("lisi", true)))

  //通过socket从nc中获取数据
  val linesDStream = ssc.socketTextStream("Hadoop01", 6666)

  /**
   * 过滤黑名单用户发言
   * zs sb sb sb sb
   * lisi fuck fuck fuck
   * jack hello
   */
  linesDStream
   .map(x => {
    val info = x.split(" ")
    (info(0), info.toList.tail.mkString(" "))
   })
   .transform(rdd => { //transform是一个RDD->RDD的操作，所以返回值必须是RDD
    /**
     * 经过leftouterjoin操作之后，产生的结果如下：
     * (zs,(sb sb sb sb),Some(true)))
     * (lisi,(fuck fuck fuck),some(true)))
     * (jack,(hello,None))
     */
    val joinRDD = rdd.leftOuterJoin(blackRDD)

    //如果是Some(true)的，说明就是黑名单用户，如果是None的，说明不在黑名单内，把非黑名单的用户保留下来
    val filterRDD = joinRDD.filter(x => x._2._2.isEmpty)

    filterRDD
   })
   .map(x=>(x._1,x._2._1)).print()

  ssc.start()
  ssc.awaitTermination()
 }
}

package StreamingDemo

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * 基于缓存的实时WordCount，在全局范围内统计单词出现次数
 */
object UpdateStateByKeyDemo {
 def main(args: Array[String]): Unit = {
  //设置日志级别
  Logger.getLogger("org").setLevel(Level.WARN)

  /**
   * 如果没有启用安全认证或者从Kerberos获取的用户为null，那么获取HADOOP_USER_NAME环境变量，
   * 并将它的值作为Hadoop执行用户设置hadoop username
   * 这里实验了一下在没有启用安全认证的情况下，就算不显式添加，也会自动获取我的用户名
   */
  //System.setProperty("HADOOP_USER_NAME","Setsuna")

  val conf = new SparkConf()
   .setAppName(this.getClass.getSimpleName)
   .setMaster("local[2]")
  val ssc = new StreamingContext(conf, Seconds(2))

  //设置Checkpoint存放的路径
  ssc.checkpoint("hdfs://Hadoop01:9000/checkpoint")

  //创建输入DStream
  val lineDStream = ssc.socketTextStream("Hadoop01", 6666)
  val wordDStream = lineDStream.flatMap(_.split(" "))
  val pairsDStream = wordDStream.map((_, 1))

  /**
   * state:代表之前的状态值
   * values：代表当前batch中key对应的values值
   */
  val resultDStream =
   pairsDStream.updateStateByKey((values: Seq[Int], state: Option[Int]) => {

    //当state为none，表示没有对这个单词做统计，则返回0值给计数器count
    var count = state.getOrElse(0)

    //遍历values，累加新出现的单词的value值
    for (value <- values) {
     count += value
    }

    //返回key对应的新state，即单词的出现次数
    Option(count)
   })

  //在控制台输出
  resultDStream.print()

  ssc.start()
  ssc.awaitTermination()
 }
}

package StreamingDemo

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * 需求：每隔2秒钟，统计最近5秒钟的搜索词中排名最靠前的3个搜索词以及出现次数
 */
object ReduceByKeyAndWindowDemo {
 def main(args: Array[String]): Unit = {

  //设置日志级别
  Logger.getLogger("org").setLevel(Level.WARN)
  //基础配置
  val conf = new SparkConf()
   .setAppName(this.getClass.getSimpleName)
   .setMaster("local[2]")

  //批处理间隔设置为1s
  val ssc = new StreamingContext(conf, Seconds(1))

  val linesDStream = ssc.socketTextStream("Hadoop01", 6666)
  linesDStream
   .flatMap(_.split(" ")) //根据空格来做分词
   .map((_, 1)) //返回(word,1)
   .reduceByKeyAndWindow(
    //定义窗口如何计算的函数
    //x代表的是聚合后的结果，y代表的是这个Key对应的下一个需要聚合的值
    (x: Int, y: Int) => x + y,
    //窗口长度为5秒
    Seconds(5),
    //窗口时间间隔为2秒
    Seconds(2)
   )
   .transform(rdd => { //transform算子对rdd做处理，转换为另一个rdd
    //根据Key的出现次数来进行排序，然后降序排列，获取最靠前的3个搜索词
    val info: Array[(String, Int)] = rdd.sortBy(_._2, false).take(3)
    //将Array转换为resultRDD
    val resultRDD = ssc.sparkContext.parallelize(info)
    resultRDD
   })
   .map(x => s"${x._1}出现的次数是：${x._2}")
   .print()

  ssc.start()
  ssc.awaitTermination()

 }
}

dstream.foreachRDD { rdd =>
  val connection=createNewConnection()
  rdd.foreach { record => connection.send(record)
  }
}

dstream.foreachRDD { rdd =>
  rdd.foreach { record =>
    val connection = createNewConnection()
    connection.send(record)
    connection.close()
  }
}

CREATE TABLE wordcount (
  word varchar(100) CHARACTER SET utf8 NOT NULL,
  count int(10) NOT NULL,
  PRIMARY KEY (word)
) ENGINE=InnoDB DEFAULT CHARSET=latin1;

package StreamingDemo

import java.sql.{Connection, DriverManager, SQLException}
import java.util

object JDBCManager {
 var connectionQue: java.util.LinkedList[Connection] = null

 /**
  * 从数据库连接池中获取连接对象
  * @return
  */
 def getConnection(): Connection = {
  synchronized({
   try {
    //如果连接池是空的，那么就实例化一个Connection类型的链表
    if (connectionQue == null) {
     connectionQue = new util.LinkedList[Connection]()
     for (i <- 0 until (10)) {
      //生成10个连接，并配置相关信息
      val connection = DriverManager.getConnection(
       "jdbc:mysql://Hadoop01:3306/test?characterEncoding=utf-8",
       "root",
       "root")
      //将连接push进连接池
      connectionQue.push(connection)
     }
    }
   } catch {
    //捕获异常并输出
    case e: SQLException => e.printStackTrace()
   }
   //如果连接池不为空，则返回表头元素，并将它在链表里删除
   return connectionQue.poll()
  })
 }

 /**
  * 当连接对象用完后，需要调用这个方法归还连接
  * @param connection
  */
 def returnConnection(connection: Connection) = {
  //插入元素
  connectionQue.push(connection)
 }

 def main(args: Array[String]): Unit = {
  //main方法测试
  getConnection()
  println(connectionQue.size())
 }
}

package StreamingDemo

import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, streaming}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object ForeachRDDDemo {
 def main(args: Array[String]): Unit = {
  //设置日志级别，避免INFO信息过多
  Logger.getLogger("org").setLevel(Level.WARN)

  //设置Hadoop的用户，不加也可以
  System.setProperty("HADOOP_USER_NAME", "Setsuna")

  //Spark基本配置
  val conf = new SparkConf()
   .setAppName(this.getClass.getSimpleName)
   .setMaster("local[2]")
  val ssc = new StreamingContext(conf, streaming.Seconds(2))

  //因为要使用updateStateByKey，所以需要使用checkpoint
  ssc.checkpoint("hdfs://Hadoop01:9000/checkpoint")

  //设置socket，跟nc配置的一样
  val linesDStream = ssc.socketTextStream("Hadoop01", 6666)
  val wordCountDStream = linesDStream
   .flatMap(_.split(" "))   //根据空格做分词
   .map((_, 1)) //生成(word,1)
   .updateStateByKey((values: Seq[Int], state: Option[Int]) => {
    //实时更新状态信息
    var count = state.getOrElse(0)
    for (value <- values) {
     count += value
    }
    Option(count)
   })

  wordCountDStream.foreachRDD(rdd => {
   if (!rdd.isEmpty()) {
    rdd.foreachPartition(part => {
     //从连接池中获取连接
     val connection = JDBCManager.getConnection()
     part.foreach(data => {
      val sql = //往wordcount表中插入wordcount信息，on duplicate key update子句是有则更新无则插入
       s"insert into wordcount (word,count) " +
        s"values ('${data._1}',${data._2}) on duplicate key update count=${data._2}"
      //使用prepareStatement来使用sql语句
      val pstmt = connection.prepareStatement(sql)
      pstmt.executeUpdate()
     })
     //在连接处提交完数据后，归还连接到连接池
     JDBCManager.returnConnection(connection)
    })
   }
  })

  ssc.start()
  ssc.awaitTermination()
 }
}