Spark中RDD算子的foreachRDD使用注意

闭包

我们的项目需要从 Kafka 消费消息，在对消息进行处理后，再写入到 ActiveMQ，以作为外部系统的数据源。基于这样的逻辑，我们就需要通过 Spark Streaming 读取 Kafka 的消息，获得的结果其实是一个 RDD。DStream 提供了foreachRDD(func)方法，通过该方法可以遍历 RDD 的每条记录，然后再通过 ActiveMQ 的 Producer 将处理后的消息发送到 ActiveMQ。

要将消息发送到 ActiveMQ，就需要建立与消息队列的连接。在传统编程实现中，最直观的做法一定是将获取连接的代码放在foreachRDD(func)方法之外，如此可以避免不必要的资源消耗与时间消耗。例如：

dstream.foreachRDD { rdd =>
  val producer = createProducer()
  rdd.foreach { message =>
    producer.send(message)
  }
}

def createProducer(): MessageProducer = {
  val conn = createActiveMQConnection()
  val session = sessionFrom(conn)
  producerFrom(session)
}

但是，这一做法在 Spark Streaming 中却行不通。原因在于：foreachRDD(func)方法中的 func 是在调用 Spark 流式计算程序的 Driver 进程中执行的，而遍历得到的 RDD 中的操作却是在 worker 中执行：

dstream.foreachRDD { rdd =>
  val producer = createProducer()  //在driver进程执行
  rdd.foreach { message =>
    producer.send(message)  //在worker进程执行
  }
}

这就需要将获得的对象（例子中包括了 Connection、Session 和 Producer）进行序列化，使其能够从 driver 发送到 worker。然而，连接等于资源相关的对象往往无法支持序列化，也无法在 worker 正确的初始化。

为了避免这种情况，一种做法是将前面的createProducer()方法搬到内部的rdd.foreach(fn)中来。可是，创建一个 connection 对象往往既费时间又费资源，针对每个 RDD 不停地创建连接，然后又关闭连接，会影响到整个系统的吞吐量和性能。

解决方案是使用foreachPartition(func)方法，通过它创建一个单独的 connection 对象，然后在 RDD 分区里使用这个连接对象将所有数据发送出去：

dstream.foreachRDD { rdd =>
  rdd.foreachPartition { partitionOfRecords =>
    sendToActiveMQ { producer =>
      partitionOfRecords.foreach(record => producer.send(record))
    }
  }
}

def sendToActiveMQ(send: MessageProducer => Unit):Unit => {
  val conn = createActiveMQConnection()
  val session = sessionFrom(conn)
  val producer = producerFrom(session)
  send(producer)
  conn.close()
}

为了避免过多的创建和释放 connection 对象，还有一个更好的方案是使用连接池。由于我在前面的代码已经将连接创建与关闭提取出专门的方法，因此只需要修改前面的sendToActiveMQ()即可：

def sendToActiveMQ(send: MessageProducer => Unit):Unit => {
  val conn = ActiveMQConnectionPool.getConnection()
  val session = sessionFrom(conn)
  val producer = producerFrom(session)
  send(producer)
  ActiveMQConnectionPool.returnConnnection(conn)
}

Spark 这种 Driver 与 Worker 互相协作的分布式架构，与单节点的编程模型存在细微差异。开发时，稍不注意就可能出现问题。当然，面对这些问题，最根本的还是要从 Spark 的设计本质来理解，问题也就迎刃而解了。

本文作者： 张逸
本文链接： zhangyi.xyz/foreachrdd-…
版权声明： 本博客所有文章除特别声明外，均采用 CC BY-NC-SA 3.0 许可协议。转载请注明出处！

单例模式

但是，以上更多的是关于匿名函数的闭包方式的讨论，第二种单例模式的处理方式与以上所述存在一些区别。
在一个 Spark 应用的执行过程中，Driver 和 Worker 是两个重要角色。Driver 程序是应用逻辑执行的起点，负责作业的调度，即 Task 任务的分发，而多个 Worker 用来管理

计算节点和创建 Executor 并行处理任务。在执行阶段，Driver 会将 Task 和 Task 所依赖的 file 和 jar 序列化后传递给对应的 Worker 机器，同时 Executor 对相应数据分区的任务进行处理。注意，所有的 Executor 上都会获取一份程序的 jar 包。

单例模式是一种常用的设计模式，但是在集群模式下的 Spark 中使用单例模式会引发一些错误。我们用下面代码作例子，解读在 Spark 中使用单例模式遇到的问题。

object Example{
  var instance:Example = new Example("default_name");
  def getInstance():Example = {
    return instance
  }
  def init(name:String){
    instance = new Example(name)
  }
}
class Example private(name1:String) extends  Serializable{
  var name = name1
}
 
object Main{
  def main(args:Array[String]) = {
    Example.init("new_name")
    val sc =  new SparkContext(newSparkConf().setAppName("test"))
 
    val rdd = sc.parallelize(1 to 10, 3)
    rdd.map(x=>{
      x + "_"+ Example.getInstance().name
    }).collect.foreach(println)
  }
}

本地运行的结果是：

1_new_name
2_new_name
3_new_name
4_new_name
7_new_name
5_new_name
8_new_name
6_new_name
10_new_name
9_new_name
12_new_name
11_new_name

注释掉 setMaster("local[5]") 部分，提交到 Spark 集群运行，得到的结果是：

1_default_name
2_default_name
3_default_name
4_default_name
5_default_name
6_default_name
7_default_name
8_default_name
9_default_name
10_default_name
11_default_name
12_default_name

注意：我们在 rdd.map 中使用了 Exampleobject，但是并没有对 Example 可序列化做任何处理，但是程序并没有抛出不可序列化异常，显而易见，当 Spark 准备闭包的时候，并没有将 Example 整合对象序列化打包传递到 worker 端执行。

这是由什么原因导致的呢？Spark 执行算子之前，会将算子需要东西准备好并打包（这就是闭包的概念），分发到不同的 executor，但这里不包括类。类存在 jar 包中，随着 jar 包分发到不同的 executors 中。当不同的 executors 执行算子需要类时，直接从分发的 jar 包取得。这时候在 driver 上对类的静态变量进行改变，并不能影响 executors 中的类。拿上面的程序做例子，jar 包存的 Example.instance = newExample("default_name")，分发到不同的 executors。这时候不同 executors 中 Example.getInstance().name 等于 "default_name"。