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.默认情况下,输出操作是一次一个执行的。它们按照它们在应用程序中定义的顺序执行。

 

分类:

技术点:

相关文章: