####好#####DStreams上的输出操作

2024-05-07 15:08
文章标签 输出 操作 #### ##### dstreams

本文主要是介绍####好#####DStreams上的输出操作，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

DStreams上的输出操作

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

Output OperationMeaning
print()在DStream的每个批数据中打印前10条元素，这个操作在开发和调试中都非常有用。在Python API中调用pprint()。
saveAsObjectFiles(prefix, [suffix])保存DStream的内容为一个序列化的文件SequenceFile。每一个批间隔的文件的文件名基于prefix和suffix生成。"prefix-TIME_IN_MS[.suffix]"，在Python API中不可用。
saveAsTextFiles(prefix, [suffix])保存DStream的内容为一个文本文件。每一个批间隔的文件的文件名基于prefix和suffix生成。"prefix-TIME_IN_MS[.suffix]"
saveAsHadoopFiles(prefix, [suffix])保存DStream的内容为一个hadoop文件。每一个批间隔的文件的文件名基于prefix和suffix生成。"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 driverrdd.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 connectionsval 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输出操作是分时执行的，它们按照应用程序的定义顺序按序执行。

这篇关于####好#####DStreams上的输出操作的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/967696。

相关文章

顺序表之创建，判满，插入，输出

顺序表之创建，判满，插入，输出

文章目录 🍊自我介绍🍊创建一个空的顺序表，为结构体在堆区分配空间🍊插入数据🍊输出数据🍊判断顺序表是否满了，满了返回值1，否则返回0🍊main函数 你的点赞评论就是对博主最大的鼓励 当然喜欢的小伙伴可以：点赞+关注+评论+收藏（一键四连）哦~ 🍊自我介绍   Hello,大家好，我是小珑也要变强（也是小珑），我是易编程·终身成长社群的一名“创始团队·嘉宾”
阅读更多...
AI(文生语音)-TTS 技术线路探索学习:从拼接式参数化方法到Tacotron端到端输出

AI(文生语音)-TTS 技术线路探索学习:从拼接式参数化方法到Tacotron端到端输出

AI(文生语音)-TTS 技术线路探索学习:从拼接式参数化方法到Tacotron端到端输出 在数字化时代，文本到语音（Text-to-Speech, TTS）技术已成为人机交互的关键桥梁，无论是为视障人士提供辅助阅读，还是为智能助手注入声音的灵魂，TTS 技术都扮演着至关重要的角色。从最初的拼接式方法到参数化技术，再到现今的深度学习解决方案，TTS 技术经历了一段长足的进步。这篇文章将带您穿越时
阅读更多...
动手学深度学习【数据操作+数据预处理】

动手学深度学习【数据操作+数据预处理】

import osos.makedirs(os.path.join('.', 'data'), exist_ok=True)data_file = os.path.join('.', 'data', 'house_tiny.csv')with open(data_file, 'w') as f:f.write('NumRooms,Alley,Price\n') # 列名f.write('NA
阅读更多...
线程的四种操作

线程的四种操作

所属专栏：Java学习        1. 线程的开启 start和run的区别： run：描述了线程要执行的任务，也可以称为线程的入口 start：调用系统函数，真正的在系统内核中创建线程（创建PCB，加入到链表中），此处的start会根据不同的系统，分别调用不同的api，创建好之后的线程，再单独去执行run（所以说，start的本质是调用系统api，系统的api
阅读更多...
Java IO 操作——个人理解

Java IO 操作——个人理解

之前一直Java的IO操作一知半解。今天看到一个便文章觉得很有道理（ 原文章），记录一下。 首先，理解Java的IO操作到底操作的什么内容，过程又是怎么样子。          数据来源的操作： 来源有文件，网络数据。使用File类和Sockets等。这里操作的是数据本身，1,0结构。    File file = new File("path");   字
阅读更多...
MySQL——表操作

MySQL——表操作

目录 一、创建表 二、查看表 2.1 查看表中某成员的数据 2.2 查看整个表中的表成员 2.3 查看创建表时的句柄 三、修改表 alter 3.1 重命名 rename 3.2 新增一列 add 3.3 更改列属性 modify 3.4 更改列名称 change 3.5 删除某列 上一篇博客介绍了库的操作，接下来来看一下表的相关操作。 一、创建表 create
阅读更多...
如何将一个文件里不包含某个字符的行输出到另一个文件？

如何将一个文件里不包含某个字符的行输出到另一个文件？

第一种： grep -v 'string' filename > newfilenamegrep -v 'string' filename >> newfilename 第二种： sed -n '/string/!'p filename > newfilenamesed -n '/string/!'p filename >> newfilename
阅读更多...
封装MySQL操作时Where条件语句的组织

封装MySQL操作时Where条件语句的组织

在对数据库进行封装的过程中，条件语句应该是相对难以处理的，毕竟条件语句太过于多样性。 条件语句大致分为以下几种： 1、单一条件，比如：where id = 1; 2、多个条件，相互间关系统一。比如：where id > 10 and age > 20 and score < 60; 3、多个条件，相互间关系不统一。比如：where (id > 10 OR age > 20) AND sco
阅读更多...
Detectorn2预训练模型复现：数据准备、训练命令、日志分析与输出目录

Detectorn2预训练模型复现：数据准备、训练命令、日志分析与输出目录

Detectorn2预训练模型复现：数据准备、训练命令、日志分析与输出目录 在深度学习项目中，目标检测是一项重要的任务。本文将详细介绍如何使用Detectron2进行目标检测模型的复现训练，涵盖训练数据准备、训练命令、训练日志分析、训练指标以及训练输出目录的各个文件及其作用。特别地，我们将演示在训练过程中出现中断后，如何使用 resume 功能继续训练，并将我们复现的模型与Model Zoo中的
阅读更多...
PHP7扩展开发之流操作

PHP7扩展开发之流操作

前言 啥是流操作？简单来讲就是对一些文件，网络的IO操作。PHP已经把这些IO操作，封装成流操作。这节，我们将使用PHP扩展实现一个目录遍历的功能。PHP示例代码如下： <?phpfunction list_dir($dir) {if (is_dir($dir) === false) {return;} $dh = opendir($dir);if ($dh == false) {ret
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2