SparkStreaming之foreachRDD

忘是亡心i 2022-06-04 08:13 164阅读 0赞

DStream中的foreachRDD是一个非常强大函数，它允许你把数据发送给外部系统。因为输出操作实际上是允许外部系统消费转换后的数据，它们触发的实际操作是DStream转换。所以要掌握它，对它要有深入了解。下面有一些常用的错误需要理解。经常写数据到外部系统需要创建一个连接的object（eg:根据TCP协议连接到远程的服务器，我们连接外部数据库需要自己的句柄）和发送数据到远程的系统为此，开发者需要在Spark的driver创建一个object用于连接。  
为了达到这个目的，开发人员可能不经意的在Spark驱动中创建一个连接对象，但是在Spark worker中 尝试调用这个连接对象保存记录到RDD中，如下：  
dstream.foreachRDD \{ rdd =>  
  val connection = createNewConnection()  // executed at the driver  
  rdd.foreach \{ record =>  
    connection.send(record) // executed at the worker  
  \}  
\}  
这是不正确的，因为这需要先序列化连接对象，然后将它从driver发送到worker中。这样的连接对象在机器之间不能  
传送。它可能表现为序列化错误（连接对象不可序列化）或者初始化错误（连接对象应该 在worker中初始化）等  
等。正确的解决办法是在worker中创建连接对象。  
然而，这会造成另外一个常见的错误-为每一个记录创建了一个连接对象。例如：  
dstream.foreachRDD \{ rdd =>  
  rdd.foreach \{ record =>  
    val connection = createNewConnection()  
    connection.send(record)  
    connection.close()  
  \}  
\}  
通常，创建一个连接对象有资源和时间的开支。因此，为每个记录创建和销毁连接对象会导致非常高的开支，明显  
的减少系统的整体吞吐量。一个更好的解决办法是利用rdd.foreachPartition方法。 为RDD的partition创建一个连接对  
象，用这个两件对象发送partition中的所有记录。  
dstream.foreachRDD \{ rdd =>  
  rdd.foreachPartition \{ partitionOfRecords =>  
    val connection = createNewConnection()  
    partitionOfRecords.foreach(record => connection.send(record))  
    connection.close()  
  \}  
\}  
最后，可以通过在多个RDD或者批数据间重用连接对象做更进一步的优化。开发者可以保有一个静态的连接对象  
池，重复使用池中的对象将多批次的RDD推送到外部系统，以进一步节省开支  
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中创建一个表用于存放数据  
mysql> create database streaming;    
Query OK, 1 row affected (0.01 sec)    
     
mysql> use streaming;    
Database changed    
mysql> show tables;    
Empty set (0.01 sec)    
     
mysql> create table wc (word varchar(20),count int(10));  
Query OK, 0 rows affected (0.05 sec)    
（2）用scala编写连接Mysql的连接池  
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(ConnectionPool.this.getClass)    
  var ds:BasicDataSource = null    
  def getDataSource=\{    
    if(ds == null)\{    
      ds = new BasicDataSource()    
      ds.setUsername("root")    
      ds.setPassword("000000")    
      ds.setUrl("jdbc:mysql://localhost:3306/streaming")    
      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程序

import org.apache.spark.SparkConf  
import org.apache.spark.streaming.\{Seconds, State, StateSpec, StreamingContext\}  
  
object ForeachRDDApp \{  
  def main(args: Array\[String\]): Unit = \{  
    val sparkConf = new SparkConf().setMaster("local\[2\]").setAppName("ForeachRDDApp")  
    val ssc = new StreamingContext(sparkConf,Seconds(10))  
  
    val lines = ssc.socketTextStream("192.168.179.160", 9999)  
    val words = lines.flatMap(\_.split(" "))  
    val pairs = words.map(word => (word, 1))  
    val result = pairs.reduceByKey(\_+\_)  
  
    result.foreachRDD( rdd =>\{  
      rdd.foreachPartition(partitionOfRecords =>\{  
        val connect = ConnectionPool.getConnection  
        connect.setAutoCommit(false)  
  
        partitionOfRecords.foreach(record =>\{  
          val sql = "insert into wc (word,count) values ('"+record.\_1+"','"+record.\_2+"')"  
          val stmt = connect.createStatement().execute(sql)  
        \})  
        connect.commit()  
      \}  
      )  
    \}  
    )  
  
    ssc.start()  
    ssc.awaitTermination()  
  \}  
  
\}