使用spark DStream的foreachRDD时要注意哪些坑?

Question

使用spark DStream的foreachRDD时要注意哪些坑

Answer 1

两个坑， 性能坑和线程坑DStream是抽象类，它把连续的数据流拆成很多的小RDD数据块， 这叫做“微批次”， spark的流式处理， 都是“微批次处理”。 DStream内部实现上有批次处理时间间隔，滑动窗口等机制来保证每个微批次的时间间隔里， 数据流以RDD的形式发送给spark做进一步处理。因此， 在一个为批次的处理时间间隔里， DStream只产生一个RDD。可以利用dstream.foreachRDD把数据发送给外部系统。 但是想要正确地， 有效率的使用它， 必须理解一下背后的机制。通常向外部系统写数据需要一个Connection对象（通过它与外部服务器交互）。程序员可能会想当然地在spark上创建一个connection对象， 然后在spark线程里用这个对象来存RDD。比如下面的程序：dstream.foreachRDD { rdd =>val connection = createNewConnection() // executed at the driverrdd.foreach { record =>connection.send(record) // executed at the worker}}这个代码会产生执行错误， 因为rdd是分布式存储的，它是一个数据结构，它是一组指向集群数据的指针， rdd.foreach会在集群里的不同机器上创建spark工作线程， 而connection对象则不会在集群里的各个机器之间传递， 所以有些spark工作线程就会产生connection对象没有被初始化的执行错误。 解决的办法可以是在spark worker里为每一个worker创建一个connection对象， 但是如果你这么做， 程序要为每一条record创建一次connection，显然效率和性能都非常差。另一种改进方法是为每个spark分区创建一个connection对象，同时维护一个全局的静态的连接迟对象， 这样就可以最好的复用connection。 另外需要注意： 虽然有多个connection对象， 但在同一时间只有一个connection.send(record)执行， 因为在同一个时间里， 只有 一个微批次的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}}有人问了个问题，为什么foreachRDD里有两层嵌套的foreach? 为什么dstream.foreachRDD里还要再套一层rdd.foreach可以这么理解, DStream.foreachRDD 是一个输出操作符，它返回的不是RDD里的一行数据， 而是输出DStream后面的RDD,在一个时间间隔里， 只返回一个RDD的“微批次”， 为了访问这个“微批次”RDD里的数据， 我们还需要在RDD数据对象上做进一步操作.。 参考下面的代码实例， 更容易理解。给顶一个 RDD [Security, Prices]数据结构dstream.foreachRDD { pricesRDD => // Loop over RDDval x= pricesRDD.countif (x > 0) // RDD has data{for(line <- pricesRDD.collect.toArray) // Look for each record in the RDD{var index = line._2.split(',').view(0).toInt // That is the indexvar timestamp = line._2.split(',').view(1).toString // This is the timestamp from sourcevar security = line._2.split(',').view(12.toString // This is the name of the securityvar price = line._2.split(',').view(3).toFloat // This is the price of the securityif (price.toFloat > 90.0){// Do something here// Sent notification, write to HDFS etc}}}}