前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >SparkStreaming之foreachRDD

SparkStreaming之foreachRDD

作者头像
全栈程序员站长
发布2022-07-02 16:45:44
3060
发布2022-07-02 16:45:44
举报

大家好,又见面了,我是你们的朋友全栈君。

首先我们来对官网的描述了解一下。

DStream中的foreachRDD是一个非常强大函数,它允许你把数据发送给外部系统。因为输出操作实际上是允许外部系统消费转换后的数据,它们触发的实际操作是DStream转换。所以要掌握它,对它要有深入了解。下面有一些常用的错误需要理解。经常写数据到外部系统需要创建一个连接的object(eg:根据TCP协议连接到远程的服务器,我们连接外部数据库需要自己的句柄)和发送数据到远程的系统为此,开发者需要在Spark的driver创建一个object用于连接。

为了达到这个目的,开发人员可能不经意的在Spark驱动中创建一个连接对象,但是在Spark worker中 尝试调用这个连接对象保存记录到RDD中,如下:

代码语言:javascript
复制
dstream.foreachRDD { 
    rdd =>
  val connection = createNewConnection()  // executed at the driver
  rdd.foreach { 
    record =>
    connection.send(record) // executed at the worker
  }
}

这是不正确的,因为这需要先序列化连接对象,然后将它从driver发送到worker中。这样的连接对象在机器之间不能

传送。它可能表现为序列化错误(连接对象不可序列化)或者初始化错误(连接对象应该 在worker中初始化)等

等。正确的解决办法是在worker中创建连接对象。

然而,这会造成另外一个常见的错误-为每一个记录创建了一个连接对象。例如:

代码语言:javascript
复制
dstream.foreachRDD { 
    rdd =>
  rdd.foreach { 
    record =>
    val connection = createNewConnection()
    connection.send(record)
    connection.close()
  }
}

通常,创建一个连接对象有资源和时间的开支。因此,为每个记录创建和销毁连接对象会导致非常高的开支,明显

的减少系统的整体吞吐量。一个更好的解决办法是利用rdd.foreachPartition方法。 为RDD的partition创建一个连接对

象,用这个两件对象发送partition中的所有记录。

代码语言:javascript
复制
dstream.foreachRDD { 
    rdd =>
  rdd.foreachPartition { 
    partitionOfRecords =>
    val connection = createNewConnection()
    partitionOfRecords.foreach(record => connection.send(record))
    connection.close()
  }
}

最后,可以通过在多个RDD或者批数据间重用连接对象做更进一步的优化。开发者可以保有一个静态的连接对象

池,重复使用池中的对象将多批次的RDD推送到外部系统,以进一步节省开支

代码语言:javascript
复制
dstream.foreachRDD { 
    rdd =>
  rdd.foreachPartition { 
    partitionOfRecords =>
    // ConnectionPool is a static, lazily initialized pool of connections
    val connection = ConnectionPool.getConnection()
    partitionOfRecords.foreach(record => connection.send(record))
    ConnectionPool.returnConnection(connection)  // return to the pool for future reuse
  }
}

需要注意的是,池中的连接对象应该根据需要延迟创建,并且在空闲一段时间后自动超时。这样就获取了最有效的

方式发生数据到外部系统。

其它需要注意的地方:

(1)输出操作通过懒执行的方式操作DStreams,正如RDD action通过懒执行的方式操作RDD。具体地看,RDD

actions和DStreams输出操作接收数据的处理。因此,如果你的应用程序没有任何输出操作或者 用于输出操作

dstream.foreachRDD(),但是没有任何RDD action操作在dstream.foreachRDD()里面,那么什么也不会执行。系统

仅仅会接收输入,然后丢弃它们。

(2)默认情况下,DStreams输出操作是分时执行的,它们按照应用程序的定义顺序按序执行。

实验1:把SparkStreaming的内部数据存入Mysql

(1)在mysql中创建一个表用于存放数据

代码语言:javascript
复制
mysql> create database sparkStreaming;
Query OK, 1 row affected (0.01 sec)
 
mysql> use sparkStreaming;
Database changed
mysql> show tables;
Empty set (0.01 sec)
 
mysql> create table searchKeyWord(insert_time date,keyword varchar(30),search_count integer);
Query OK, 0 rows affected (0.05 sec)

(2)用scala编写连接Mysql的连接池

代码语言:javascript
复制
import java.sql.Connection
import java.sql.PreparedStatement
import java.sql.ResultSet

import org.apache.commons.dbcp.BasicDataSource
import org.apache.log4j.Logger

object scalaConnectPool {
  val  log = Logger.getLogger(scalaConnectPool.this.getClass)
  var ds:BasicDataSource = null
  def getDataSource={
    if(ds == null){
      ds = new BasicDataSource()
      ds.setUsername("root")
      ds.setPassword("iamhaoren")
      ds.setUrl("jdbc:mysql://localhost:3306/sparkStreaming")
      ds.setDriverClassName("com.mysql.jdbc.Driver")
      ds.setInitialSize(20)
      ds.setMaxActive(100)
      ds.setMinIdle(50)
      ds.setMaxIdle(100)
      ds.setMaxWait(1000)
      ds.setMinEvictableIdleTimeMillis(5*60*1000)
      ds.setTimeBetweenEvictionRunsMillis(10*60*1000)
      ds.setTestOnBorrow(true)
    }
    ds
  }

  def getConnection : Connection= {
    var connect:Connection = null
    try {
      if(ds != null){
        connect = ds.getConnection
      }else{
        connect = getDataSource.getConnection
      }
    }
    connect
  }

  def shutDownDataSource: Unit=if (ds !=null){ds.close()}

  def closeConnection(rs:ResultSet,ps:PreparedStatement,connect:Connection): Unit ={
    if(rs != null){rs.close}
    if(ps != null){ps.close}
    if(connect != null){connect.close}
  }

}

(3)编写SparkStreaming程序

代码语言:javascript
复制
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

object dataToMySQL {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("use the foreachRDD write data to mysql").setMaster("local[2]")
    val ssc = new StreamingContext(conf,Seconds(10))

    val streamData = ssc.socketTextStream("master",9999)
    val wordCount = streamData.map(line =>(line.split(",")(0),1)).reduceByKeyAndWindow(_+_,Seconds(60))
    val hottestWord = wordCount.transform(itemRDD => {
      val top3 = itemRDD.map(pair => (pair._2, pair._1))
        .sortByKey(false).map(pair => (pair._2, pair._1)).take(3)
      ssc.sparkContext.makeRDD(top3)
    })


    hottestWord.foreachRDD( rdd =>{
      rdd.foreachPartition(partitionOfRecords =>{
        val connect = scalaConnectPool.getConnection
        connect.setAutoCommit(false)
        val stmt = connect.createStatement()
        partitionOfRecords.foreach(record =>{
          stmt.addBatch("insert into searchKeyWord (insert_time,keyword,search_count) values (now(),'"+record._1+"','"+record._2+"')")
        })
        stmt.executeBatch()
        connect.commit()
      }
      )
    }
    )



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

(4)编写一个socket端的数据模拟器

代码语言:javascript
复制
import java.io.{PrintWriter}  
import java.net.ServerSocket  
import scala.io.Source  
  
  
object streamingSimulation {  
  def index(n: Int) = scala.util.Random.nextInt(n)  
  
  def main(args: Array[String]) {  
    // 调用该模拟器需要三个参数,分为为文件路径、端口号和间隔时间(单位:毫秒)  
    if (args.length != 3) {  
      System.err.println("Usage: <filename> <port> <millisecond>")  
      System.exit(1)  
    }  
  
    // 获取指定文件总的行数  
    val filename = args(0)  
    val lines = Source.fromFile(filename).getLines.toList  
    val filerow = lines.length  
  
    // 指定监听某端口,当外部程序请求时建立连接  
    val listener = new ServerSocket(args(1).toInt)  
  
    while (true) {  
      val socket = listener.accept()  
      new Thread() {  
        override def run = {  
          println("Got client connected from: " + socket.getInetAddress)  
          val out = new PrintWriter(socket.getOutputStream(), true)  
          while (true) {  
            Thread.sleep(args(2).toLong)  
            // 当该端口接受请求时,随机获取某行数据发送给对方  
            val content = lines(index(filerow))  
            println("-------------------------------------------")  
            println(s"Time: ${System.currentTimeMillis()}")  
            println("-------------------------------------------")  
            println(content)  
            out.write(content + '\n')  
            out.flush()  
          }  
          socket.close()  
        }  
      }.start()  
    }  
  }  
  
}  

实验数据为:

spark Streaming better than storm you need it yes do it

(5)实验启动

在客户端启动数据流模拟

对socket端的数据模拟器程序进行 jar文件的打包,并放入到集群目录中

启动程序如下:

代码语言:javascript
复制
java -cp DataSimulation.jar streamingSimulation /root/application/upload/Information 9999 1000

启动SparkStreaming程序

结果如下:

SparkStreaming之foreachRDD
SparkStreaming之foreachRDD

发布者:全栈程序员栈长,转载请注明出处:https://javaforall.cn/148544.html原文链接:https://javaforall.cn

本文参与 腾讯云自媒体同步曝光计划,分享自作者个人站点/博客。
如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
相关产品与服务
文件存储
文件存储(Cloud File Storage,CFS)为您提供安全可靠、可扩展的共享文件存储服务。文件存储可与腾讯云服务器、容器服务、批量计算等服务搭配使用,为多个计算节点提供容量和性能可弹性扩展的高性能共享存储。腾讯云文件存储的管理界面简单、易使用,可实现对现有应用的无缝集成;按实际用量付费,为您节约成本,简化 IT 运维工作。
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档