限制 apache spark 作业运行时间答案

【问题标题】：Limit apache spark job running duration限制 apache spark 作业运行时间
【发布时间】：2018-02-03 04:53:51
【问题描述】：

我想在具有超时参数的集群环境中提交作业，是否有办法让 spark 杀死一个正在运行的作业，如果它超过了允许的持续时间？

【问题讨论】：

取决于您希望何时开始超时。您可以在驱动程序中自己执行此操作。将来有一个超时开始工作。当超时到期时，驱动程序将关闭，从而终止 Spark 作业。
我通过 rest 提交作业，并在请求中传递属性，所以我想我需要在那里配置它。

标签： apache-spark

【解决方案1】：

在 Spark 2.1.0 中，没有内置解决方案（添加的一个非常好的功能！）。

您可以使用speculation 功能重新启动长任务和spark.task.maxFailures 来杀死太多重新启动的任务。

但这绝对不干净，Spark 缺少一个真正的“断路器”来停止长时间的任务（例如菜鸟SELECT * FROM DB）

另一方面，您可以使用 Spark Web UI Web API：

1) 获取正在运行的作业：GET http://SPARK_CLUSTER_PROD/api/v1/applications/application_1502112083252_1942/jobs?status=running

（这将为您提供一个包含 submissionTime 字段的数组，您可以使用它来查找长期工作）

2) 终止作业：POST http://SPARK_CLUSTER_PROD/stages/stage/kill/?id=23881&terminate=true 用于每个作业阶段。

我相信Spark也有隐藏API，你可以试试。

【讨论】：

【解决方案2】：

您可以使用 YARN REST api 从您的服务中终止 spark 应用程序。我正在使用以下代码来停止长时间运行的 spark 应用程序。以下代码使用 httpclient 库。

 def killApplication(applicationId: String) : Boolean = {
val appKillPut = new HttpPut(s"http://xx.xx.xx.xx:8088//ws/v1/cluster/apps/$applicationId/state")
val json = new JSONObject(Map("state"-> "KILLED"))

val params = new StringEntity(json.toString(),"UTF-8")
params.setContentType("application/json")

appKillPut.addHeader("Content-Type", "application/json")
appKillPut.addHeader("Accept", "*/*")
appKillPut.setEntity(params)

println(s"Request payload ${json.toString}")

val client: CloseableHttpClient = HttpClientBuilder.create().build()
val response: CloseableHttpResponse = client.execute(appKillPut)
val responseBody = EntityUtils.toString(response.getEntity)
println(s"Response payload ${responseBody}")
val statusCode: Int = response.getStatusLine.getStatusCode
if(statusCode == 200 || statusCode == 201 || statusCode == 202) {
  println(s"Successfully stopped the application : ${applicationId}")
  true
} else {
  false
}
}

希望这会有所帮助。

拉维

【讨论】：