spark Dstreams-数据源

IT那活儿 2022-03-22

639

点击上方“IT那活儿”，关注后了解更多内容，不管IT什么活儿，干就完了！！！

01

基本数据源

1.1 文件数据源

Spark Streaming会监控你指定的目录下的任何文件。

一个目录会被监控，像hdfs://namenode:8040/logs，目录中的所有文件会被发现并处理。
全局模式应用，像hdfs://namenode:8040/logs/2017/*，这样spark streaming会监控所有匹配的目录下的文件。
所有文件必须是相同格式文件。
文件会被当作时间段的一部分根据修改时间而不是创建时间。
文件被读取处理后，如果对文件又进行了修改，不会导致spark streaming重新读取文件，即忽略更新。
就算没有修改任何文件，目录下文件数量越多，扫描的时间就越长。
如果使用通配符标识监控目录，重命名目录名称以匹配通配符，只有目录中修改时间当前窗口内的文件才会包含在流中。
调用FileSystem.setTimes()来修复时间戳是在以后的窗口中摄取文件的一种方法，尽管其内容未修改。

1.2 对象存储数据源

文件系统（像HDFS）倾向于在创建输出流后立即对其文件设置修改时间，当文件打开时，数据被完全写入之前，他可能会被包括在数据流中，之后再对文件的更新将被忽略，也就是说更新可能会丢失数据。

要保证在窗口中获取更新，请将文件写入未被监控的目录，然后再输出流关闭后立即将其重名为目标目录，则将摄取新数据。

对比之下，对象存储（例如S3，Azure）通常具有缓慢的重命名操作，因为数据实际上是复制的，此外，重命名的对象可能将rename()操作的时间作为其修改时间，因此可能不会被视为原始创建时间所暗示的窗口的一部分。

需要对目标对象存储进行仔细测试，以验证存储的时间戳行为是否与saprk streaming所预期的一致，直接写入目标目录可能是通过所选对象存储进行数据流传输的合适策略。

1.3 通过RDD创建DStream

为了用于测试，用户可以通过一系列RDD创建DStream，每一个加入到队列中的RDD将会被看成是DStream的a batch。

02

高级数据源

类源需要与外部非Spark库接口，其中一些库具有复杂的依赖关系（例如Kafka）。

因此，为了尽量减少与依赖项的版本冲突相关的问题，从这些源创建数据流的功能已转移到单独的库中，必要时可以显式链接到这些库。

请注意，Spark shell中没有这些高级源，因此无法在shell中测试基于这些高级源的应用程序。如果您真的想在Spark shell中使用它们，那么您必须下载相应的Maven工件的JAR及其依赖项，并将其添加到类路径中。

03

自定义数据源

可以从自定义数据源中创建输入数据流。您所要做的就是实现一个用户定义的接收器（receiver ），它可以从自定义源接收数据并将其推入Spark。

3.1 实现自定义接收器

自定义接收器必须实现Receiver类并实现其两个方法：

onStart()：开始接收数据需要做的事情；
onStop()：停止接收数据需要做的事情。

onStart()、onStop()都不能无限期阻塞，通常的，onStart()将启动负责接收数据的线程，onStop()将确保停止接收数据的线程，接收数据线程还可以使用isStopped()来检查他们是否应该停止接收数据。

一旦接收到数据就可以调用store()方法将数据存储在spark中，store(data)是Receiver提供的一种方法，store（）有多种风格，允许一次存储一条接收到的数据记录，或作为对象/序列化字节的整个集合。

请注意，用于实现接收器的store（）的风格会影响其可靠性和容错语义。应捕获并正确处理接收线程中的任何异常，以避免接收器的无声故障。restart（<exception>）将通过异步调用onStop（）然后在延迟后调用onStart（）来重新启动接收器。stop（<exception>）将调用onStop（）并终止接收器。

此外，reportError（<error>）会向驱动程序报告错误消息（在日志和UI中可见），而无需停止/重新启动接收器。

以下是通过套接字接收文本流的自定义接收器。它将文本流中的“\n”分隔行视为记录，并使用Spark存储它们。如果接收线程在连接或接收时出错，则重新启动接收器以再次尝试连接。

在spark streaming中使用自定义Receiver：

3.2 接收器可靠性

正如Spark Streaming编程指南中简要讨论的，基于可靠性和容错语义，有两种接收器：

可靠接收器-对于允许确认发送数据的可靠源，可靠接收器正确地向源确认数据已可靠地接收并存储在Spark中（即成功复制）。通常，实现此接收器需要仔细考虑源确认的语义。
不可靠接收器-不可靠接收器不向源发送确认。

要实现可靠的接收器，必须使用存储（多条记录）来存储数据。这种风格的存储是一个阻塞调用，只有在所有给定的记录都存储在Spark中之后才会返回。如果接收器配置的存储级别使用复制（默认情况下已启用），则此调用将在复制完成后返回。因此，它确保数据被可靠地存储，并且接收器现在可以适当地确认源。这确保当接收器在复制数据中间失败时，没有数据丢失——缓冲的数据将不被确认。

不可靠的接收器不必实现任何这种逻辑。它可以简单地从源接收记录，并使用store（single record）一次插入一条记录。虽然它没有得到存储（多个记录）的可靠性保证，但是更容易实现。

下表总结了这两种接收机的特性：