Scala中基于Hive列值循环调用curl的性能问题

我来帮你拆解下这段代码的性能瓶颈，然后给出几个针对性的优化方案，让你的任务能高效运行：

先说说原代码的核心问题 #

1. 单点串行处理：用collect()把Hive表的所有数据拉到Driver节点，然后单线程循环执行curl。如果数据量大，不仅Driver会面临内存溢出风险，串行执行的效率也极低，完全没用到Spark集群的分布式能力。
2. 不必要的进程开销：每次调用curl都会启动一个新的系统进程，进程的创建和销毁会带来额外的性能损耗。
3. 字符串拼接错误：原代码里的JSON请求体拼接有语法问题，"{\"searchText\":\"+'z'+",\"qf\":\"applId\"}"没办法正确把变量z插入到JSON里，这会导致请求失败。

优化方案一：利用Spark分布式并行处理 #

核心思路是不要把数据拉到Driver，而是让Executor节点并行处理各自分区的数据，把请求压力分散到集群各个节点上。

import scala.concurrent.{Future, Await}
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global

// 保留DataFrame，不要用collect()拉取到Driver
val df = hiveContext.sql("select application_number from t")

// 封装发送请求的逻辑
def sendHttpReq(appNumber: String): Unit = {
  // 用字符串插值正确生成JSON请求体
  val jsonBody = s"""{"searchText":"$appNumber","qf":"applId"}"""
  val cmd = Seq(
    "curl", "-X", "POST", "--insecure",
    "--header", "Content-Type: application/json",
    "--header", "Accept: application/json",
    "-d", jsonBody,
    "http://your-target-api-url" // 替换成实际的目标URL
  )
  
  // 执行命令并处理结果（可选：记录日志、处理失败）
  val process = cmd.run()
  val exitCode = process.exitValue()
  if (exitCode != 0) {
    println(s"Request failed for app number: $appNumber, exit code: $exitCode")
  }
}

// 用foreachPartition让每个Executor处理自己的分区数据
df.foreachPartition { partition =>
  // 每个分区内用Future并行发送请求，提升处理速度
  val requestFutures = partition.map(row => {
    val appNum = row.getString(0)
    Future { sendHttpReq(appNum) }
  })
  
  // 等待当前分区的所有请求完成，再处理下一个分区
  Await.result(Future.sequence(requestFutures), 15.minutes) // 根据实际情况调整超时时间
}

这个方案的优势：

• 分布式处理，充分利用Spark集群的资源，避免Driver单点瓶颈
• 分区内并行请求，比串行执行效率提升数倍

优化方案二：批量发送请求（性能提升最显著） #

如果你的目标API支持批量查询，那一定要用这个方案——把多个application_number打包成一个请求发送，能大幅减少HTTP请求的总次数，降低TCP握手、HTTP头部等开销。

假设目标API接受批量参数（比如JSON数组），代码可以改成这样：

df.foreachPartition { partition =>
  // 把当前分区的所有application_number收集成列表
  val appNumbers = partition.map(_.getString(0)).filter(_ != null).toList
  
  if (appNumbers.nonEmpty) {
    // 生成批量请求的JSON体
    val jsonBody = s"""{"searchTexts":[${appNumbers.map(n => s""""$n"""").mkString(",")}],"qf":"applId"}"""
    val cmd = Seq(
      "curl", "-X", "POST", "--insecure",
      "--header", "Content-Type: application/json",
      "--header", "Accept: application/json",
      "-d", jsonBody,
      "http://your-target-api-url"
    )
    
    val process = cmd.run()
    val exitCode = process.exitValue()
    // 这里可以解析批量响应，处理每个appNumber的结果
    if (exitCode != 0) {
      println(s"Batch request failed for ${appNumbers.size} entries")
    }
  }
}

优化方案三：替换curl为JVM HTTP客户端（减少进程开销） #

每次启动curl进程都会带来额外的系统资源消耗，换成Scala/Java的HTTP客户端库（比如requests-scala、akka-http），直接在JVM内发送请求，能显著提升性能，还更容易处理重试、超时等逻辑。

以requests-scala为例：
首先在sbt里添加依赖：

libraryDependencies += "com.lihaoyi" %% "requests" % "0.8.0"

然后修改代码：

import requests._
import scala.concurrent.{Future, Await}
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global

val df = hiveContext.sql("select application_number from t")

def sendHttpReq(appNumber: String): Unit = {
  try {
    val response = requests.post(
      url = "http://your-target-api-url",
      headers = Map("Content-Type" -> "application/json", "Accept" -> "application/json"),
      json = Map("searchText" -> appNumber, "qf" -> "applId"),
      verifySsl = false // 对应curl的--insecure参数
    )
    
    if (response.statusCode != 200) {
      println(s"Request failed for $appNumber: ${response.statusCode}, response: ${response.text()}")
    }
  } catch {
    case e: Exception => println(s"Exception occurred for $appNumber: ${e.getMessage}")
  }
}

df.foreachPartition { partition =>
  val futures = partition.map(row => Future { sendHttpReq(row.getString(0)) })
  Await.result(Future.sequence(futures), 15.minutes)
}

额外的优化建议 #

• 并发控制：不要无限制并行发送请求，避免压垮目标服务。可以自定义线程池控制并发数：

  import java.util.concurrent.Executors
  import scala.concurrent.ExecutionContext
  
  val customEC = ExecutionContext.fromExecutor(Executors.newFixedThreadPool(8)) // 限制并发数为8
  

  然后在创建Future时指定这个ExecutionContext：Future { ... }(customEC)
• 错误重试：添加重试机制，比如针对网络波动导致的请求失败，自动重试2-3次
• 分区调整：如果数据量极大，可以用df.repartition(20)调整分区数，让每个分区的数据量更合理，避免单个分区处理过多数据导致超时
• 日志记录：添加详细的日志（比如用log4j），方便排查请求失败的问题

内容的提问来源于stack exchange，提问作者user6325753