REST API

Flink 具有监控 API ，可用于查询正在运行的作业以及最近完成的作业的状态和统计信息。该监控 API 被用于 Flink 自己的仪表盘，同时也可用于自定义监控工具。

该监控 API 是 REST-ful 风格的，可以接受 HTTP 请求并返回 JSON 格式的数据。

API reference

• /cluster：Returns all cluster data sets

• /jars：Returns a list of all jars previously uploaded via '/jars/upload'

• /jobs：Returns an overview over all jobs and their current state

• /jobs/overview：Returns an overview over all jobs

• /jobs/:jobid：Returns details of a job

• /jobs/:jobid/exceptions：Returns the most recent exceptions that have been handled by Flink for this job

全部api可查看REST API | Apache Flink

测试

/jars

public CloseableHttpResponse getJobsList() throws URISyntaxException, IOException {
        HttpGet httpGet = new HttpGet();
        httpGet.setURI(new URI(baseURI + "/jobs"));
        return httpClient.execute(httpGet);
}

/jobs/:jobid

public CloseableHttpResponse getJobDetail(String jobId) throws URISyntaxException, IOException {
        HttpGet httpGet = new HttpGet();
        httpGet.setURI(new URI(baseURI + String.format("/jobs/%s", jobId)));
        return httpClient.execute(httpGet);
    }

/jobs/:jobid/exceptions

public CloseableHttpResponse getJobError(String jobId) throws URISyntaxException, IOException {
        HttpGet httpGet = new HttpGet();
        httpGet.setURI(new URI(baseURI + String.format("/jobs/%s/exceptions", jobId)));
        return httpClient.execute(httpGet);
    }

public static void main(String[] args) throws IOException, URISyntaxException {
        HttpClientUtil httpClientUtil=new HttpClientUtil("xxx:8081");
        System.out.println("---------jars---------");
        CloseableHttpResponse jobsList = httpClientUtil.showJars();
        HttpEntity entity = jobsList.getEntity();
        System.out.println(EntityUtils.toString(entity,"utf-8"));
        System.out.println("---------jobs---------");
        CloseableHttpResponse jobsInfo = httpClientUtil.getJobsList();
        HttpEntity entityInfo = jobsInfo.getEntity();
        System.out.println(EntityUtils.toString(entityInfo,"utf-8"));
        System.out.println("----------------------");
        CloseableHttpResponse jobsError = httpClientUtil.getJobError("9807ceb347deea2d58cfec08401aeea1");
        HttpEntity entityError = jobsError.getEntity();
        System.out.println(EntityUtils.toString(entityError,"utf-8"));
    }

测试结果