从 Spark 调用休息服务

Question

我正在尝试找出从 Spark 调用 Rest 端点的最佳方法。

我目前的方法（解决方案 [1]）看起来像这样 -

val df = ... // some dataframe

val repartitionedDf = df.repartition(numberPartitions)

lazy val restEndPoint = new restEndPointCaller() // lazy evaluation of the object which creates the connection to REST. lazy vals are also initialized once per JVM (executor)

val enrichedDf = repartitionedDf 
.map(rec => restEndPoint.getResponse(rec)) // calls the rest endpoint for every record
.toDF

我知道我可以使用 .mapPartitions() 而不是 .map()，但是查看 DAG，看起来 spark 优化了重新分区 -> 无论如何映射到 mapPartition。

在第二种方法（解决方案 [2]）中，为每个分区创建一次连接，并为分区内的所有记录重复使用。

  val newDs = myDs.mapPartitions(partition => {
  val restEndPoint = new restEndPointCaller /*creates a db connection per partition*/

  val newPartition = partition.map(record => {
    restEndPoint.getResponse(record, connection)
  }).toList // consumes the iterator, thus calls readMatchingFromDB 

  restEndPoint.close() // close dbconnection here
  newPartition.iterator // create a new iterator
})

在这第三种方法（解决方案 [3]）中，每个 JVM（执行程序）创建一次连接，并在执行程序处理的所有分区中重复使用。

    lazy val connection = new DbConnection /*creates a db connection per partition*/
    val newDs = myDs.mapPartitions(partition => {

          val newPartition = partition.map(record => {
            readMatchingFromDB(record, connection)
          }).toList // consumes the iterator, thus calls readMatchingFromDB 
        
          newPartition.iterator // create a new iterator

        })
    connection.close() // close dbconnection here

[a] 对于非常相似的解决方案 [1] 和 [3]，我对 lazy val 如何工作的理解是否正确？目的是将每个执行程序/JVM 的连接数限制为 1，并重用打开的连接来处理后续请求。我会为每个 JVM 创建 1 个连接还是每个分区创建 1 个连接？

[b] 还有其他方法可以控制我们向其余端点发出的请求数 (RPS) 吗？

[c] 如果有更好、更有效的方法，请告诉我。

谢谢！

Answer 1

IMO 使用mapPartitions 的第二种解决方案更好。首先，你明确地告诉你期望实现什么。转换的名称和实现的逻辑非常清楚地说明了这一点。对于第一个选项，您需要了解 Apache Spark 如何优化处理。现在对您来说可能很明显，但您还应该考虑将在 6 个月、1 年、2 年等等之后为您的代码工作的人，或者只是考虑您。他们应该比repartition + map 更了解mapPartitions。

此外，使用 map 进行重新分区的优化可能会在内部发生变化（我不相信，但您仍然可以认为这是一个有效点），此时您的工作会表现得更差。

最后，使用第二种解决方案，您可以避免在序列化过程中可能遇到的许多问题。在您编写的代码中，驱动程序将创建端点对象的一个​​实例，将其序列化并发送给执行程序。所以是的，也许它会是一个单一的实例，但前提是它是可序列化的。

---

[编辑] 感谢您的澄清。您可以通过不同的方式实现您的目标。要让每个 JVM 恰好有 1 个连接，您可以使用一种称为单例的设计模式。在 Scala 中，它很容易表达为 object（我在 Google https://alvinalexander.com/scala/how-to-implement-singleton-pattern-in-scala-with-object 上找到的第一个链接）

而且它非常好，因为您不需要序列化任何东西。单例直接从执行器端的类路径中读取。有了它，您肯定会拥有给定对象的一个​​实例。

[a] 与解决方案 [1] 和 [3] 非常相似，是我的 了解lazy val 如何正确工作？目的是 将每个执行程序/JVM 的连接数限制为 1 并重用 用于处理后续请求的打开连接。我会不会 每个 JVM 或每个分区创建 1 个连接？ 它将为每个分区创建 1 个连接。你可以执行这个小测试来看看：

  class SerializationProblemsTest extends FlatSpec   {
    val conf = new SparkConf().setAppName("Spark serialization problems test").setMaster("local") 
    val sparkContext = SparkContext.getOrCreate(conf)
    "lazy object" should "be created once per partition" in {
      lazy val restEndpoint = new NotSerializableRest()
      sparkContext.parallelize(0 to 120).repartition(12)
        .mapPartitions(numbers => {
           //val restEndpoint = new NotSerializableRest()
           numbers.map(nr => restEndpoint.enrich(nr))
      })
      .collect()
   }
 }  
 class NotSerializableRest() {
   println("Creating REST instance")
   def enrich(id: Int): String = s"${id}"
}

它应该打印 Creating REST instance 12 次（分区数）

[b] 有什么方法可以控制请求数 (RPS) 我们到其余端点？

要控制请求的数量，您可以使用类似于数据库连接池的方法：HTTP 连接池（一个快速找到的链接：HTTP connection pooling using HttpClient）。

但也许另一种有效的方法是处理较小的数据子集？因此，您可以将其拆分为不同的较小的微批次（如果它是流式作业），而不是处理 30000 行。它应该给你的网络服务更多的“休息”。

否则，您也可以尝试发送批量请求（Elasticsearch 会同时索引/删除多个文档https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html）。但这取决于网络服务是否允许您这样做。