Spark官方3 ---------Spark Streaming编程指南（1.5.0）

时间：2021-05-13 15:00:37

Design Patterns for using foreachRDD

dstream.foreachRDD是一个强大的原语，允许将数据发送到外部系统。然而，了解如何正确有效地使用该原语很重要。避免一些常见的错误如下。

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

dstream.foreachRDD { rdd =>

  val connection = createNewConnection()  // executed at the driver

  rdd.foreach { record =>

    connection.send(record) // executed at the worker

  }

}

这是不正确的，因为这需要将连接对象序列化并从驱动程序发送给worker。这样的连接对象很少能跨机器传输。此错误可能会显示为序列化错误（连接对象不可序列化），初始化错误（连接对象需要在工作人员初始化）等。正确的解决方案是在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 /batches 之间重复使用连接对象来进一步优化。可以维护连接对象的静态池，而不是多个批次的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

  }

}

请注意，池中的连接应根据需要懒惰创建，如果不使用一段时间，则会超时。这实现了最有效地将数据发送到外部系统。

其他要记住的要点

1.DStreams通过输出操作进行延迟执行，就像RDD由RDD actions懒惰执行一样。具体来说，DStream输出操作中的RDD动作强制处理接收到的数据。因此，如果您的应用程序没有任何输出操作，或者在dstream.foreachRDD（）中没有任何RDD action操作，那么任何操作都将不会被执行。系统将简单地接收数据并将其丢弃。

2.默认情况下，输出操作是一次一个执行的。它们按照它们在应用程序中定义的顺序执行。

相关文章