Spark Streaming算子开发实例

Spark Streaming算子开发实例

transform算子开发

transform操作应用在DStream上时,可以用于执行任意的RDD到RDD的转换操作,还可以用于实现DStream API中所没有提供的操作,比如说,DStreamAPI中并没有提供将一个DStream中的每个batch,与一个特定的RDD进行join的操作,DStream中的join算子只能join其他DStream,但是我们自己就可以使用transform操作来实现该功能。

实例:黑名单用户实时过滤

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()
 }
}

测试

启动nc,传入用户及其发言信息

可以看到程序实时的过滤掉了在黑名单里的用户发言

updateStateByKey算子开发

updateStateByKey算子可以保持任意状态,同时不断有新的信息进行更新,这个算子可以为每个key维护一份state,并持续不断的更新state。对于每个batch来说,Spark都会为每个之前已经存在的key去应用一次State更新函数,无论这个key在batch中是否有新的值,如果State更新函数返回的值是none,那么这个key对应的state就会被删除;对于新出现的key也会执行state更新函数。

要使用该算子,必须进行两个步骤

  • 定义state——state可以是任意的数据类型
  • 定义state更新函数——用一个函数指定如何使用之前的状态,以及从输入流中获取新值更新状态

注意:updateStateByKey操作,要求必须开启Checkpoint机制

实例:基于缓存的实时WordCount

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()
 }
}

测试

开启nc,输入单词

控制台实时输出的结果

window滑动窗口算子开发

Spark Streaming提供了滑动窗口操作的支持,可以对一个滑动窗口内的数据执行计算操作
在滑动窗口中,包含批处理间隔、窗口间隔、滑动间隔

  • 对于窗口操作而言,在其窗口内部会有N个批处理数据
  • 批处理数据的大小由窗口间隔决定,而窗口间隔指的就是窗口的持续时间,也就是窗口的长度
  • 滑动时间间隔指的是经过多长时间窗口滑动一次,形成新的窗口,滑动间隔默认情况下和批处理时间间隔的相同

注意:滑动时间间隔和窗口时间间隔的大小一定得设置为批处理间隔的整数倍

用一个官方的图来作为说明

批处理间隔是1个时间单位,窗口间隔是3个时间单位,滑动间隔是2个时间单位。对于初始的窗口time1-time3,只有窗口间隔满足了才触发数据的处理。所以滑动窗口操作都必须指定两个参数,窗口长度和滑动时间间隔。在Spark Streaming中对滑动窗口的支持是比Storm更加完善的。

Window滑动算子操作

算子 描述
window() 对每个滑动窗口的数据执行自定义的计算
countByWindow() 对每个滑动窗口的数据执行count操作
reduceByWindow() 对每个滑动窗口的数据执行reduce操作
reduceByKeyAndWindow() 对每个滑动窗口的数据执行reduceByKey操作
countByValueAndWindow() 对每个滑动窗口的数据执行countByValue操作

reduceByKeyAndWindow算子开发

实例:在线热点搜索词实时滑动统计

每隔2秒钟,统计最近5秒钟的搜索词中排名最靠前的3个搜索词以及出现次数

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 Output操作概览

Spark Streaming允许DStream的数据输出到外部系统,DSteram中的所有计算,都是由output操作触发的,foreachRDD输出操作,也必须在里面对RDD执行action操作,才能触发对每一个batch的计算逻辑。

转换 描述
print() 在Driver中打印出DStream中数据的前10个元素。主要用于测试,或者是不需要执行什么output操作时,用于简单触发一下job。
saveAsTextFiles(prefix,
[suffix])
将DStream中的内容以文本的形式保存为文本文件,其中每次批处理间隔内产生的文件以prefix-TIME_IN_MS[.suffix]的方式命名。
saveAsObjectFiles(prefix
, [suffix])
将DStream中的内容按对象序列化并且以SequenceFile的格式保存。其中每次批处理间隔内产生的文件以prefix-TIME_IN_MS[.suffix]的方式命名。
saveAsHadoopFiles(pref
ix, [suffix])
将DStream中的内容以文本的形式保存为Hadoop文件,其中每次批处理间隔内产生的文件
以prefix-TIME_IN_MS[.suffix]的方式命名。
foreachRDD(func) 最基本的输出操作,将func函数应用于DStream中的RDD上,这个操作会输出数据到外部系
统,比如保存RDD到文件或者网络数据库等。需要注意的是func函数是在运行该streaming
应用的Driver进程里执行的。

foreachRDD算子开发

foreachRDD是最常用的output操作,可以遍历DStream中的每个产生的RDD并进行处理,然后将每个RDD中的数据写入外部存储,如文件、数据库、缓存等,通常在其中针对RDD执行action操作,比如foreach

使用foreachRDD操作数据库

通常在foreachRDD中都会创建一个Connection,比如JDBC Connection,然后通过Connection将数据写入外部存储

误区一:在RDD的foreach操作外部创建Connection

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

这种方式是错误的,这样的方式会导致Connection对象被序列化后被传输到每一个task上,但是Connection对象是不支持序列化的,所以也就无法被传输

误区二:在RDD的foreach操作内部创建Connection

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

这种方式虽然是可以的,但是执行效率会很低,因为它会导致对RDD中的每一条数据都创建一个Connection对象,通常Connection对象的创建都是很消耗性能的

合理的方式

  • 第一种:使用RDD的foreachPartition操作,并且在该操作内部创建Connection对象,这样就相当于为RDD的每个partition创建一个Connection对象,节省了很多资源
  • 第二种:自己手动封装一个静态连接池,使用RDD的foreachPartition操作,并且在该操作内部从静态连接池中,通过静态方法获取到一个连接,连接使用完之后再放回连接池中。这样的话,可以在多个RDD的partition之间复用连接了

实例:实时全局统计WordCount,并将结果保存到MySQL数据库中

MySQL数据库建表语句如下

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

在IDEA中添加mysql-connector-java-5.1.40-bin.jar

代码如下

连接池的代码,其实一开始有想过用静态块来写个池子直接获取,但是如果考虑到池子宽度不够用的问题,这样的方式其实更好,一开始,实例化一个连接池出来,被调用获取连接,当连接全部都被获取了的时候,池子空了,就再实例化一个池子出来

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())
 }
}

wordcount代码

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()
 }
}

打开nc,输入数据

在另一个终端对wordcount的结果进行查询,可以发现是实时发生变化的

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持我们。#niming{position:fixed; bottom:0; z-index:9999;right:0;width:100%;background:#e5e5e5;}.fengyu{width:99%;padding:0.5%;text-align:center;}a{text-decoration: none;color:#000;}.tuiguangbiaoji{font-size:10px;margin:0px 10px;}

(0)

相关推荐

  • Spark学习笔记Spark Streaming的使用

    1. Spark Streaming Spark Streaming是一个基于Spark Core之上的实时计算框架,可以从很多数据源消费数据并对数据进行处理 Spark Streaing中有一个最基本的抽象叫DStream(代理),本质上就是一系列连续的RDD,DStream其实就是对RDD的封装 DStream可以认为是一个RDD的工厂,该DStream里面生产都是相同业务逻辑的RDD,只不过是RDD里面要读取数据的不相同 在一个批次的处理时间间隔里, DStream只产生一个RDD DSt

  • Spark Streaming算子开发实例

    Spark Streaming算子开发实例 transform算子开发 transform操作应用在DStream上时,可以用于执行任意的RDD到RDD的转换操作,还可以用于实现DStream API中所没有提供的操作,比如说,DStreamAPI中并没有提供将一个DStream中的每个batch,与一个特定的RDD进行join的操作,DStream中的join算子只能join其他DStream,但是我们自己就可以使用transform操作来实现该功能. 实例:黑名单用户实时过滤 package

  • JQuery EasyUI 结合ztrIee的后台页面开发实例

    JQuery EasyUI 结合 zTree树形结构制作web页面.easyui用起来比较简单,很好的封装了jquery的部分功能,使用起来更加方便,但是从1.2.3版本以后,商业用途是需要付费的, zTree是国内的大牛们搞的一个jquery树形tree插件,感觉很好用,很强大,而且完全免费,API等做的也非常不错.推荐 easyui 是一个基于 jQuery 的框架,集成了各种用户界面插件. easyui 提供建立现代化的具有交互性的 javascript 应用的必要的功能. 使用 easy

  • Vue.js组件使用开发实例教程

    组件 组件可以扩展HTML元素,封装可重用的代码,在较高的层面上,组件是自定义元素,vue.js的编译器为它添加特殊功能,在有些情况下,组件也可以是原生HTML元素的形式,以is特性扩展. Vue.js的组件可以理解为预先定义好了行为的ViewModel类.一个组件可以预定义很多选项,但最核心的是以下几个: 模板(template):模板声明了数据和最终展现给用户的DOM之间的映射关系. 初始数据(data):一个组件的初始数据状态.对于可复用的组件来说,这通常是私有的状态. 接受的外部参数(p

  • Android中的指纹识别demo开发实例

    指纹识别是在Android 6.0之后新增的功能,因此在使用的时候需要先判断用户手机的系统版本是否支持指纹识别.另外,实际开发场景中,使用指纹的主要场景有两种: 纯本地使用.即用户在本地完成指纹识别后,不需要将指纹的相关信息给后台. 与后台交互.用户在本地完成指纹识别后,需要将指纹相关的信息传给后台. 由于使用指纹识别功能需要一个加密对象(CryptoObject)该对象一般是由对称加密或者非对称加密获得.上述两种开发场景的实现大同小异,主要区别在于加密过程中密钥的创建和使用,一般来说,纯本地的

  • asp.net网络数据库开发实例精解 源文件

    出版社:清华大学出版社本书特点:通过大量实例,将ASP.NET的技术点贯穿起来深入剖析网络数据库开发的技术点和难点贯穿了作者多年的开发经验,具有较高的参考价值专门设计了综合案例,让您领会项目开发的思路代码规范,层次清楚,注释丰富,易于理解配有多媒体演示光盘,加速学习.读者对象:ASP.NET网络数据库开发人员大中专院校相关专业的学生毕业设计的学生ASP.NET培训人员asp.net网络数据库开发实例精解 源文件下载后用虚拟光驱软件打开,即可,最全的文件了,整个光盘的内容都在里面了,jb51.ne

  • Android USB转串口通信开发实例详解

     Android USB转串口通信开发实例详解 好久没有写文章了,年前公司新开了一个项目,是和usb转串口通信相关的,需求是用安卓平板通过usb转接后与好几个外设进行通信,一直忙到最近,才慢慢闲下来,趁着这个周末不忙,记录下usb转串口通信开发的基本流程. 我们开发使用的是usb主机模式,即:安卓平板作为主机,usb外设作为从机进行数据通信.整个开发流程可以总结为以下几点: 1.发现设备 UsbManager usbManager = (UsbManager) context.getSystem

  • Android miniTwitter登录界面开发实例

    本文要演示的Android开发实例是如何完成一个Android中的miniTwitter登录界面,下面将分步骤讲解怎样实现图中的界面效果,让大家都能轻松的做出美观的登录界面. 先贴上最终要完成的效果图: miniTwitter登录界面的布局分析 首先由界面图分析布局,基本可以分为三个部分,下面分别讲解每个部分. 第一部分是一个带渐变色背景的LinearLayout布局,关于背景渐变色就不再贴代码了,效果如下图所示: 第二部分,红色线区域内,包括1,2,3,4 如图所示: 红色的1表示的是一个带圆

  • Android自定义TitleView标题开发实例

    Android开发过程中,经常遇到一个项目需要重复的定义相同样式的标题栏,Android相继推出了actionBar, toolBar, 相信有用到的朋友也会遇到一些不如意的时候,比如标题栏居中时,需要自定义xml文件给toolBar等,不了解actionBar,toolBar的可以去找相应的文章了解,这里介绍自定义titleBar满足国内主题风格样式的情况. 为了提前看到效果,先上效果图: 前期准备 1.为标题栏titleView预定义id,在values下的ids.xml中 <?xml ve

  • Android 音乐播放器的开发实例详解

    本文将引导大家做一个音乐播放器,在做这个Android开发实例的过程中,能够帮助大家进一步熟悉和掌握学过的ListView和其他一些组件.为了有更好的学习效果,其中很多功能我们手动实现,例如音乐播放的快进快退等. 先欣赏下本实例完成后运行的界面效果: 首先我们建立项目,我使用的SDK是Android2.2的,然后在XML中进行布局. 上方是一个ListView用来显示我们的音乐列表,中间是一个SeekBar可以拖动当前音乐的播放进度,之所以用SeekBar而不用ProgressBar是因为我们需

  • Android6.0指纹识别开发实例详解

    Android6.0指纹识别开发实例详解 最近在做android指纹相关的功能,谷歌在android6.0及以上版本对指纹识别进行了官方支持.当时在FingerprintManager和FingerprintManagerCompat这两个之间纠结,其中使用FingerprintManager要引入com.android.support:appcompat-v7包,考虑到包的大小,决定使用v4兼容包FingerprintManagerCompat来实现. 主要实现的工具类FingerprintU

随机推荐