Spark系列--Spark Streaming(七)DStreams输出

青旅半醒 2022-05-18 08:12 252阅读 0赞

## 前言 ##

输出操作指定了对流数据经转化操作得到的数据所要执行的操作(例如把结果推入外部数据库或输出到屏幕上)。与 RDD 中的惰性求值类似，如果一个 DStream 及其派生出的 DStream 都没有被执行输出操作，那么这些 DStream 就都不会被求值。如果 StreamingContext 中没有设定输出操作，整个 context 就都不会启动。

<table> 
 <thead> 
  <tr> 
   <th>Tables</th> 
   <th align="center">Are</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Output Operation</td> 
   <td align="center">Meaning</td> 
  </tr> 
  <tr> 
   <td>print()</td> 
   <td align="center">在运行流程序的驱动结点上打印DStream中每一批次数据的最开始10个元素。这用于开发和调试。在Python API中，同样的操作叫pprint()。</td> 
  </tr> 
  <tr> 
   <td>saveAsTextFiles(prefix, [suffix])</td> 
   <td align="center">以text文件形式存储这个DStream的内容。每一批次的存储文件名基于参数中的prefix和suffix。”prefix-Time_IN_MS[.suffix]”.</td> 
  </tr> 
  <tr> 
   <td>saveAsObjectFiles(prefix, [suffix])</td> 
   <td align="center">以Java对象序列化的方式将Stream中的数据保存为 SequenceFiles . 每一批次的存储文件名基于参数中的为”prefix-TIME_IN_MS[.suffix]”. Python中目前不可用。</td> 
  </tr> 
  <tr> 
   <td>saveAsHadoopFiles(prefix, [suffix])</td> 
   <td align="center">将Stream中的数据保存为 Hadoop files. 每一批次的存储文件名基于参数中的为”prefix-TIME_IN_MS[.suffix]”. Python API Python中目前不可用。</td> 
  </tr> 
  <tr> 
   <td>foreachRDD(func)</td> 
   <td align="center">这是最通用的输出操作，即将函数func用于产生于stream的每一个RDD。其中参数传入的函数func应该实现将每一个RDD中数据推送到外部系统，如将RDD存入文件或者通过网络将其写入数据库。注意：函数func在运行流应用的驱动中被执行，同时其中一般函数RDD操作从而强制其对于流RDD的运算。</td> 
  </tr> 
 </tbody> 
</table>

通用的输出操作 foreachRDD()，它用来对 DStream 中的 RDD 运行任意计算。这和transform() 有些类似，都可以让我们访问任意 RDD。在 foreachRDD() 中，可以重用我们在 Spark 中实现的所有行动操作。比如，常见的用例之一是把数据写到诸如 MySQL 的外部数据库中。

**需要注意的：**  
1）连接不能写在driver层面  
2）如果写在foreach则每个RDD都创建，得不偿失  
3）增加foreachPartition，在分区创建  
4）可以考虑使用连接池优化

dstream.foreachRDD { rdd =>
       // executed at the driver 序列化错误
       // error val connection = createNewConnection() 
    
      rdd.foreachPartition { partitionOfRecords =>
        // ConnectionPool is a static, lazily initialized pool of connections
        val connection = ConnectionPool.getConnection()
        // executed at the worker
        partitionOfRecords.foreach(record => connection.send(record) 
        )
    
        // return to the pool for future reuse
        ConnectionPool.returnConnection(connection)  
      }
    }