简单的flink on yarn rest api

素颜马尾好姑娘i 2022-11-19 12:43 246阅读 0赞

直接上代码吧

package yarn;
    
    import com.alibaba.fastjson.JSON;
    import com.alibaba.fastjson.JSONObject;
    
    import com.mzlion.easyokhttp.HttpClient;
    
    import org.apache.commons.collections.CollectionUtils;
    
    
    import java.util.ArrayList;
    import java.util.Arrays;
    
    import java.util.List;
    
    import java.util.concurrent.TimeUnit;
    
    public class GetYarnApps {
    //    private static volatile ConcurrentHashMap<String, String> currentTaskMap = new ConcurrentHashMap<String, String>();
        private static volatile List<String> failedTaskList =  new ArrayList<String>();
    
    
        public static void main(String[] args) throws InterruptedException {
    //        String url = "http://xx:8088/ws/v1/cluster/apps";
    //        String url = "http://xx:8088/ws/v1/cluster/apps?queue=users"; //todo 指定队列
            //todo 生产环境
            String url = "http://xx:8088/ws/v1/cluster/apps?queue=flink"; //todo 指定队列
            String jobUrl= "http://xx:8088/proxy/application_1609329247342_52776/jobs";
            String perfer_url = "http://xx:8088/proxy/";
    
            String url2 = "http://xx:8088/ws/v1/cluster/apps/application_1326821518301_0005"; //todo 查看指定任务
            String url3 = "http://xx:8088/ws/v1/cluster/apps/application_1326821518301_0005/state"; //todo 查看指定任务状态
            String url4 = "http://xx:8088/ws/v1/cluster/metrics"; //todo 整个集群指标
    //        String url="http://xx:8088/cluster/apps/FINISHED";
    //        String url="http://xx:7180/static/apidocs/";
            String responseData = HttpClient.get(url).asString();
            JSONObject jsonObject = JSONObject.parseObject(responseData);
            String app = jsonObject.getJSONObject("apps").getString("app");
            while (true) {
                failedTaskList.clear();
                List<YarnApp> appList = JSON.parseArray(app, YarnApp.class);
                for (int i = 0; i < appList.size(); i++) {
                    String queue = appList.get(i).getQueue();
                    String id = appList.get(i).getId();
    
                    String state = appList.get(i).getState();
    
                    String name = appList.get(i).getName();
    
    //                System.out.println("任务状态state = " + state);
                    System.out.println("yarn任务名称name = " + name+",___,"+"任务state = " + state+",___,"+"任务Id = " + id);
    
                    String jobs_url = perfer_url+id+"/jobs";
                    String job_responseData = HttpClient.get(jobs_url).asString();
                    JSONObject job_json = JSONObject.parseObject(job_responseData);
    
                     String job_id = job_json.getJSONArray("jobs").getJSONObject(0).getString("id");
                    String job_status = job_json.getJSONArray("jobs").getJSONObject(0).getString("status");
                    System.out.println("job_id = " + job_id+",________,job_status = "+job_status);
                    System.out.println();
                }
    
                TimeUnit.SECONDS.sleep(30);
    
            }
    
         }
    
        //todo 循环对比目前的任务是否存在。
        public static Boolean RunningTaskList(){
            //todo 在这里可以获取mysql或者redis当前我们提交的任务。
            //todo 这里的案例是从redis获取,或者本地写死
    //        private static volatile List<String> failedTaskList =  Collections.synchronizedList(new ArrayList<String>());
            List<String> list1 = Arrays.asList("Zeppelin Flink on yarn");
    //        List<String> list2= Arrays.asList("Zeppelin Flink on yarn");
            boolean equalCollection = CollectionUtils.isEqualCollection(list1, failedTaskList);
    
            return equalCollection;
        }
    
    
    
    }

结果为：

yarn任务名称name = logtail\_to\_ods,\_\_\_,任务state = RUNNING,\_\_\_,任务Id = application\_1609329247342\_52776  
job\_id = df86bec2aaad6625e13579eab6143e85,\_\_\_\_\_\_\_\_,job\_status = RUNNING

yarn任务名称name = ods\_to\_dwd,\_\_\_,任务state = RUNNING,\_\_\_,任务Id = application\_1609329247342\_52786  
job\_id = 67e21be3af197c6e94a8fa9faca03571,\_\_\_\_\_\_\_\_,job\_status = RUNNING

yarn任务名称name = core\_streamload\_task,\_\_\_,任务state = RUNNING,\_\_\_,任务Id = application\_1609329247342\_52790  
job\_id = 2847ec4ce265ba73aa260b439689343c,\_\_\_\_\_\_\_\_,job\_status = RUNNING

yarn任务名称name = his\_streamload,\_\_\_,任务state = RUNNING,\_\_\_,任务Id = application\_1609329247342\_52795  
job\_id = 64a5ae6e4e1c95b148dbdb5563c1ce83,\_\_\_\_\_\_\_\_,job\_status = RUNNING

需要的pom依赖：

<dependency>
        <groupId>com.mzlion</groupId>
        <artifactId>easy-okhttp</artifactId>
        <version>1.1.0</version>
        <scope>provided</scope>
    
    </dependency>

<dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-yarn_${scala.binary.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>