如何在Spark Scala代码中实现外部JAR执行完成后再运行后续步骤

问题根因 #

该问题本质是InvokeTakaraJar方法内部调用外部JAR时使用了异步执行逻辑，没有阻塞Driver的当前线程，才会导致后续读取表的步骤提前触发。

解决方案 #

方案1：修改InvokeTakaraJar内部逻辑，阻塞等待JAR执行完成 #

绝大多数Java/Scala调用外部JAR采用Runtime.getRuntime.exec()或ProcessBuilder实现，默认不会阻塞当前线程，你需要主动调用进程的waitFor()方法等待JAR执行完全结束，示例代码如下：

def InvokeTakaraJar(params: Array[String]): Int = {
  // 构造外部JAR启动进程
  val process = new ProcessBuilder("java", "-jar", "你的业务JAR包路径", params:_*).start()
  // 可选：打印JAR运行的标准输出和错误输出，方便排查执行问题
  scala.io.Source.fromInputStream(process.getInputStream).getLines.foreach(println)
  scala.io.Source.fromInputStream(process.getErrorStream).getLines.foreach(println)
  // 阻塞当前Driver线程，直到JAR进程执行结束，返回进程退出码
  val exitCode = process.waitFor()
  // 非0退出码可主动抛出异常，中断后续Spark流程
  if(exitCode != 0) throw new RuntimeException(s"外部JAR执行失败，退出码：$exitCode")
  exitCode
}

只要在InvokeTakaraJar中添加了waitFor()逻辑，Driver线程就会等JAR完全执行完成，再运行后续的GetDBTable步骤。

方案2：无法修改InvokeTakaraJar实现时，新增表更新校验逻辑 #

如果InvokeTakaraJar是第三方封装的方法无法修改，可以在调用完该方法后新增轮询校验逻辑，确认表更新完成后再执行步骤2：

1. 提前给业务逻辑加更新完成标识：比如给目标业务表加最后更新时间字段，或者新增一个独立的标记表专门记录JAR更新完成状态
2. 调用完InvokeTakaraJar后轮询校验标识，确认更新完成再读表，示例如下：

val jarStartTime = new java.sql.Timestamp(System.currentTimeMillis())
InvokeTakaraJar(parameter)

// 轮询配置：最长等待30分钟，每10秒校验一次
val maxWaitMs = 30 * 60 * 1000
val intervalMs = 10 * 1000
var waitedMs = 0L
var updateFinished = false

while(waitedMs < maxWaitMs && !updateFinished) {
  // 执行SQL校验更新是否完成，此处可替换为你自己的校验逻辑
  val lastUpdateTime = spark.sql("select max(update_time) from 你的目标表").head().getTimestamp(0)
  if(lastUpdateTime != null && lastUpdateTime.after(jarStartTime)) {
    updateFinished = true
  } else {
    Thread.sleep(intervalMs)
    waitedMs += intervalMs
  }
}

if(!updateFinished) throw new RuntimeException("等待外部JAR更新表超时，终止后续流程")
// 确认更新完成后再执行读表逻辑
var df = GetDBTable(parameter)

注意事项 #

• 不要把InvokeTakaraJar的调用放到RDD/DataSet的算子内部，这类算子的逻辑会分发到Executor端执行，不仅会导致JAR被重复调用，也会出现时序问题，你已经确认该调用仅在Driver端运行即可规避该问题。
• 如果InvokeTakaraJar是调用同进程内的Java类方法而非启动独立JAR进程，出现异步问题大概率是方法内部开启了独立线程未等待执行结束，这种情况需要拿到异步线程对应的Future对象，调用get()方法阻塞等待线程执行完成即可。

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