SparkStreaming之数据输出

忘是亡心i 2023-02-25 08:44 2阅读 0赞

## 一.简介 ##

输出操作允许将DStream的数据推出到外部系统，例如数据库或文件系统。由于输出操作实际上允许外部系统使用转换后的数据，因此它们会触发所有DStream转换的实际执行【类似于RDD的操作】。

## 二.算子 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70]  
翻译：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70 1]

## 三.foreachRDD使用详解 ##

dstream.foreachRDD是一个强大的原语，可以将数据发送到外部系统。但是，重要的是要了解如何正确有效地使用此原语。应避免的一些常见错误。

通常，将数据写入外部系统需要创建一个连接对象（例如，到远程服务器的TCP连接），并使用该对象将数据发送到远程系统。为此，开发人员可能会无意间尝试在Spark驱动程序中创建连接对象，然后尝试在Spark辅助程序中使用该对象以将记录保存在RDD中。

dstream.foreachRDD {  rdd =>
      val connection = createNewConnection()  // executed at the driver
      rdd.foreach {  record =>
        connection.send(record) // executed at the worker
      }
    }

这是不正确的，因为这需要将连接对象序列化并从驱动程序发送给工作程序。这样的连接对象很少能在机器之间转移。此错误可能表现为序列化错误【**连接对象不可序列化**】，初始化错误【**连接对象需要在工作程序中初始化**】等。正确的解决方案是在工作程序中创建连接对象。

但是，这可能会导致另一个常见错误，为每个记录创建一个新的连接。

dstream.foreachRDD {  rdd =>
      rdd.foreach {  record =>
        val connection = createNewConnection()
        connection.send(record)
        connection.close()
      }
    }

通常，创建连接对象会浪费时间和资源。因此，为每个记录创建和销毁连接对象会导致不必要的高开销，并且会大大降低系统的整体吞吐量。更好的解决方案是使用 rdd.foreachPartition来创建单个连接对象，并使用该连接在RDD分区中发送所有记录。

dstream.foreachRDD {  rdd =>
      rdd.foreachPartition {  partitionOfRecords =>
        val connection = createNewConnection()
        partitionOfRecords.foreach(record => connection.send(record))
        connection.close()
      }
    }

这将分摊许多记录上的连接创建开销。

最后，可以通过在多个RDD/Partition之间重用连接对象来进一步优化。与将多个批次的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
      }
    }

请注意，应按需延迟创建池中的连接，如果一段时间不使用，则超时。这样可以最有效地将数据发送到外部系统。

其他要记住的要点：

*  DStream由输出操作延迟执行，就像RDD由RDD操作延迟执行一样。具体来说，DStream输出操作内部的RDD动作会强制处理接收到的数据。因此，如果应用程序没有任何输出操作，或者dstream.foreachRDD()内部没有任何RDD操作，就不会执行任何输出操作。系统将仅接收数据并将其丢弃。
 *  默认情况下，输出操作一次执行一次。它们按照在应用程序中定义的顺序执行。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70]: /images/20230209/4f66a7e75f3b468fa4b1af8aaecb71d5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI3NjAwNzIz_size_16_color_FFFFFF_t_70 1]: /images/20230209/f653055560514d71aae912bb866c1654.png