尝试使用以下 rest API 远程运行 flink 作业,但它的抛出错误
curl -X POST -H 'Content-Type: application/json' --data '
{
"type": "object",
"id": "urn:jsonschema:org:apache:flink:runtime:webmonitor:handlers:JarRunRequestBody",
"properties": {
"programArgsList" : {
"type" : "array",
"items" : [
"input-kafka-server": "****",
"input-kafka-topics": "****",
"input-job-name": "****"
}
}
}
' http://x.x.x.x:8081/jars/810ac968-5d5f-450d-aafc-22655238d617.jar/run
{"errors":["Request 不匹配预期的格式 JarRunRequestBody."]}
在此处找到的 jar 运行处理程序的请求正文的说明是 jar 运行消息的 JSON 架构规范。您需要在请求中指定的只是properties
,而不是"type"
和"id"
字段。因此,您的请求应如下所示:
curl -X POST -H 'Content-Type: application/json' --data '
{
"programArgsList" : [
"--input-kafka-server",
"value-input-kafka-server",
"--input-kafka-topics",
"value-kafka-topics",
"--input-job-name",
"value-job-name"
],
"parallelism": 30
}
' http://x.x.x.x:8081/jars/810ac968-5d5f-450d-aafc-22655238d617.jar/run
如果有人正在寻找通过 REST API 提交 Flink 作业的 Python 代码。
import json
url = base_url + "/jars/" + jar_id + "/run"
myheader = {'content-type': 'application/json'}
mydata = {
"programArgsList": [
"--input-kafka-server", "value-input-kafka-server",
"--input-kafka-topics", "value-kafka-topics",
"--input-job-name","value-job-name"
],
"parallelism": 30
}
x = requests.post(url, data=json.dumps(mydata), headers=myheader)