登录 |  注册
首页 >  云计算&大数据 >  Spark >  DStream的输出操作

DStream的输出操作

DStreams上的输出操作

输出操作允许DStream的操作推到如数据库、文件系统等外部系统中。因为输出操作实际上是允许外部系统消费转换后的数据,它们触发的实际操作是DStream转换。目前,定义了下面几种输出操作:

Output Operation Meaning
print() 在DStream的每个批数据中打印前10条元素,这个操作在开发和调试中都非常有用。在Python API中调用pprint()
saveAsObjectFiles(prefix, [suffix]) 保存DStream的内容为一个序列化的文件SequenceFile。每一个批间隔的文件的文件名基于prefixsuffix生成。"prefix-TIME_IN_MS[.suffix]",在Python API中不可用。
saveAsTextFiles(prefix, [suffix]) 保存DStream的内容为一个文本文件。每一个批间隔的文件的文件名基于prefixsuffix生成。"prefix-TIME_IN_MS[.suffix]"
saveAsHadoopFiles(prefix, [suffix]) 保存DStream的内容为一个hadoop文件。每一个批间隔的文件的文件名基于prefixsuffix生成。"prefix-TIME_IN_MS[.suffix]",在Python API中不可用。
foreachRDD(func) 在从流中生成的每个RDD上应用函数func的最通用的输出操作。这个函数应该推送每个RDD的数据到外部系统,例如保存RDD到文件或者通过网络写到数据库中。需要注意的是,func函数在驱动程序中执行,并且一般都有RDD action在里面推动RDD流的计算。

利用foreachRDD的设计模式

dstream.foreachRDD是一个强大的原语,发送数据到外部系统中。然而,明白怎样正确地、有效地用这个原语是非常重要的。下面几点介绍了如何避免通常错误。

  • 经常写数据到外部系统需要建一个连接对象(例如到远程服务器的TCP连接),用它发送数据到远程系统。为了达到这个目的,开发人员可能不经意的在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()
      })
  })

这就将连接对象的创建开销分摊到了partition的所有记录上了。

  • 最后,可以通过在多个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
      })
  })

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

其它需要注意的地方:

  • 输出操作通过懒执行的方式操作DStreams,正如RDD action通过懒执行的方式操作RDD。具体地看,RDD actions和DStreams输出操作接收数据的处理。因此,如果你的应用程序没有任何输出操作或者用于输出操作dstream.foreachRDD(),但是没有任何RDD action操作在dstream.foreachRDD()里面,那么什么也不会执行。系统仅仅会接收输入,然后丢弃它们。
  • 默认情况下,DStreams输出操作是分时执行的,它们按照应用程序的定义顺序按序执行。
上一篇: DStream中的转换
下一篇: DStreams缓存或持久化
推荐文章
  • MD5(Message-DigestAlgorithm5)是一种广泛使用的散列函数(哈希函数),由美国密码学家罗纳德·李维斯特(RonaldL.Rivest)在1991年设计。MD5的作用是对任意长度的信息生成一个固定长度(128位,即32个十六进制字符)的“指纹”或“消息摘要”,并且几乎不可能找到
  • 循环冗余校验(CyclicRedundancyCheck,CRC)是一种用于检测数据传输和存储过程中发生错误的技术,属于一种基于数学原理的错误检测编码(ErrorDetectionCoding)方法。它通过在原始数据上附加一个固定长度的校验码,使得接收端可以通过同样的计算规则对收到的数据进行校验,确
  • AES(AdvancedEncryptionStandard)是一种广泛使用的对称密钥加密算法,它是美国国家标准与技术研究院(NIST)于2001年制定的加密标准,用于替代原有的DES(DataEncryptionStandard)。AES算法以其高效性、安全性和可靠性而著称,在众多应用领域中被广泛
  • RSA(Rivest-Shamir-Adleman)是一种广泛应用的非对称加密算法,由RonRivest、AdiShamir和LenAdleman在1977年提出。其安全性基于数学上的大数因子分解难题,即对于足够大的两个素数p和q而言,已知它们的乘积很容易,但想要从这个乘积中恢复原始的素数则异常困难
  • 最小生成树(MinimumSpanningTree,MST)是一种图论算法,用于在一个带权重的无向连通图中找到一棵包括所有顶点且总权重尽可能小的树。常见的最小生成树算法有两种:Prim算法和Kruskal算法。Prim算法原理:Prim算法是一种贪心算法,它从图中的一个顶点开始,逐步增加边,每次都添
  • 关于最短路径算法的Java实现,这里简述一下几种常用的算法及其基本原理,并给出一个Dijkstra算法的基本实现框架。Dijkstra算法(适用于无负权边的图)Dijkstra算法用于寻找图中一个顶点到其他所有顶点的最短路径。它维护了一个距离表,用来存储从源点到各个顶点的已知最短距离,并且每次都会选
学习大纲