Flink 有了一些查询job状态指标的API,这些监控 API is a REST-ful API 接受 HTTP 请求并返回JSON data ；这些监控API以jobManager中web server 为基础，默认其监听端口为8081，当然你可以在flink-conf.yaml 配置中通过 rest.port 自定义其端口，下面我们就对其REST-ful API 进行尝试。
一： Flink 上传JAR包REST-ful API
请求方式：POST
请求路径：http://localhost:8081/jar/upload
请求头： application/x-java-archive
POST http://localhost:8081/jars/upload
说明 The jar must be sent as multi-part data. Make sure that the "Content-Type" header is set to "application/x-java-archive"
另外一个方式，可以尝试用下面请求提交jar包
curl -X POST --header "Except:" -F "jarfile=@/home/opt/dbappLot-flink-1.0-SNAPSHOT-jar-with-dependencies.jar" http://localhost:8081/jars/upload
其返回值：
{"filename":"/tmp/flink-web-26298886-4da6-4bf4-b635-fa301dee092e/flink-web-upload/82ea69b3-27c5-469d-992a-82be8c3d929d_dbappLot-flink-1.0-SNAPSHOT-jar-with-dependencies.jar","status":"success"}
其中filename 为存在地址
其中82ea69b3-27c5-469d-992a-82be8c3d929d_dbappLot-flink-1.0-SNAPSHOT-jar-with-dependencies.jar 为jarid
二：查看JAR包ID REST-ful API
请求 GET http://localhost:8081/jars
返回值：
{
"address": "http://localhost:8081",
"files": [
{
"id": "82ea69b3-27c5-469d-992a-82be8c3d929d_dbappLot-flink-1.0-SNAPSHOT-jar-with-dependencies.jar",
"name": "dbappLot-flink-1.0-SNAPSHOT-jar-with-dependencies.jar",
"uploaded": 1617180292000,
"entry": [
{
"name": "com.dbapp.main.ApplicationMain",
"description": null
}
]
}
]
}
三：启动JOB REST-ful API
请求方式：POST
请求路径：http://localhost:8081/jar/{jarId}/run
请求 POST http://localhost:8081/jars/82ea69b3-27c5-469d-992a-82be8c3d929d_dbappLot-flink-1.0-SNAPSHOT-jar-with-dependencies.jar/run
在请求中可以带上参数
allowNonRestoredState (optional): Boolean value that specifies whether the job submission should be rejected if the savepoint contains state that cannot be mapped back to the job.
savepointPath (optional): String value that specifies the path of the savepoint to restore the job from.
programArg (optional): Comma-separated list of program arguments.
entry-class (optional): String value that specifies the fully qualified name of the entry point class. Overrides the class defined in the jar file manifest.
parallelism (optional): Positive integer value that specifies the desired parallelism for the job.
POST http://localhost:8081/jars/82ea69b3-27c5-469d-992a-82be8c3d929d_dbappLot-flink-1.0-SNAPSHOT-jar-with-dependencies.jar/run? parallelism =2
1、http://192.168.196.3:8081/jars/a9ffd519-5222-40aa-9228-22464417a055_examples-java-1.0.jar/run?entry-class=io.github.streamingwithflink.demo.BatchCountFromJdbc&programArgsList=--bootstrap.servers hadoop2.test.yunwei.puppet.dh:6667,hadoop3.test.yunwei.puppet.dh:6667,hadoop4.test.yunwei.puppet.dh:6667 --group.id m3gcn_tab_map_race_3 --client.id m3gcn_tab_map_race_3 --config.file hdfs://DHTestCluster/user/gezhihui/flink/configuration/m3gcn_tab_map_race2.txt --job.name m3gcn_tab_map_race
2、http://localhost:8081/jars/java-3.0.jar/run
{
"entryClass":"streamingwithflink.ApplicationMain"
"programArgsList": [
"--name",
"wen"]
}
请求体中的programArgsList可以传值到main函数的string[]args，取值可以通过ParameterTool parameters =
ParameterTool.fromArgs(args);String name = parameters.get("name");所以他的key-value的格式是key为"--key"或者"-key",
value为key的下一行，根据上述代码，value不要以"-"作为开头，这里也可以自己去解析String[]，这里args就对应我们programArgsList
的传参，如3所示
3、http://localhost:8081/jars/java-2.0.jar/run
{
"entryClass":"streamingwithflink.ApplicationMain"
"programArgsList": [
"--name",
"wen"]
}
其放回值为：
{
"jobid": "c18ac90bea03abcbc117a85a1080ba3b"
}
四：查看 JOB REST-ful API
请求方式：GET
请求路径：http://localhost:8081/jobs
返回值：{"jobs":[{"id":"c18ac90bea03abcbc117a85a1080ba3b","status":"RUNNING"},{"id":"43948e17785acc47deb6344d9165c6b3","status":"CANCELED"}]}
五：获取JOB详细信息 REST-ful API
请求方式：GET
请求路径：http://localhost:8081/jobs/{jobId}
返回值：
{
"jid": "c18ac90bea03abcbc117a85a1080ba3b",
"name": "jobName",
"isStoppable": false,
"state": "RUNNING",
"start-time": 1617189444432,
"end-time": -1,
"duration": 374118,
"now": 1617189818550,
"timestamps": {
"RESTARTING": 0,
"FINISHED": 0,
"CANCELED": 0,
"FAILING": 0,
"RECONCILING": 0,
"RUNNING": 1617189444440,
"SUSPENDED": 0,
"FAILED": 0,
"CANCELLING": 0,
"CREATED": 1617189444432
},
"vertices": [
{
"id": "cbc357ccb763df2852fee8c4fc7d55f2",
"name": "Source: dbappLot_kafkaSource -> flattenJson -> avitorMap -> schemaMap -> unFlattenJson -> (Sink: Print to Std. Out, Sink: KafkaSink)",
"parallelism": 1,
"status": "RUNNING",
"start-time": 1617189444517,
"end-time": -1,
"duration": 374033,
"tasks": {
"CANCELED": 0,
"SCHEDULED": 0,
"RUNNING": 1,
"FINISHED": 0,
"RECONCILING": 0,
"CREATED": 0,
"FAILED": 0,
"DEPLOYING": 0,
"CANCELING": 0
},
"metrics": {
"read-bytes": 0,
"read-bytes-complete": false,
"write-bytes": 0,
"write-bytes-complete": false,
"read-records": 0,
"read-records-complete": false,
"write-records": 0,
"write-records-complete": false
}
}
],
"status-counts": {
"CANCELED": 0,
"SCHEDULED": 0,
"RUNNING": 1,
"FINISHED": 0,
"RECONCILING": 0,
"CREATED": 0,
"FAILED": 0,
"DEPLOYING": 0,
"CANCELING": 0
},
"plan": {
"jid": "c18ac90bea03abcbc117a85a1080ba3b",
"name": "jobName",
"nodes": [
{
"id": "cbc357ccb763df2852fee8c4fc7d55f2",
"parallelism": 1,
"operator": "",
"operator_strategy": "",
"description": "Source: dbappLot_kafkaSource -> flattenJson -> avitorMap -> schemaMap -> unFlattenJson -> (Sink: Print to Std. Out, Sink: KafkaSink)",
"optimizer_properties": {}
}
]
}
}
六：关闭JOB REST-ful API
请求方式：PATCH
请求路径：http://localhost:8081/jobs/{jobId}
返回值：
