lanyuanxiaoyao/hudi-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feature(yarn-query): 优化 yarn-query 为集群共享查询

Browse Source 
原本一个 yarn-query 对应一个集群，集群多了之后，比较浪费，改为一个组件可以查询任意集群的信息，减少部署的复杂性
This commit is contained in:
兰缘小妖
2023-06-02 09:09:26 +08:00
parent 7768b427aa
commit cd676367c6
22 changed files with 112 additions and 491 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
service-web/src/main/java/com/lanyuanxiaoyao/service/web/controller/YarnController.java
View File

@@ -7,7 +7,7 @@ import com.lanyuanxiaoyao.service.configuration.entity.AmisResponse;
import com.lanyuanxiaoyao.service.configuration.entity.yarn.YarnApplication;
import com.lanyuanxiaoyao.service.configuration.entity.yarn.YarnQueue;
import com.lanyuanxiaoyao.service.configuration.entity.yarn.YarnRootQueue;
import com.lanyuanxiaoyao.service.forest.service.YarnClusterService;
import com.lanyuanxiaoyao.service.forest.service.YarnService;
import com.lanyuanxiaoyao.service.web.entity.YarnApplicationVO;
import com.lanyuanxiaoyao.service.web.entity.YarnClusterVO;
import com.lanyuanxiaoyao.service.web.utils.ComparatorUtil;
@@ -45,10 +45,11 @@ public class YarnController extends BaseController {
"finishedTime",
YarnApplication::getFinishedTime
);
private final YarnClusterService yarnClusterService;
private final YarnService yarnService;
public YarnController(YarnClusterService yarnClusterService) {
this.yarnClusterService = yarnClusterService;
@SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
public YarnController(YarnService yarnService) {
this.yarnService = yarnService;
}
@GetMapping("job_list")
@@ -69,8 +70,9 @@ public class YarnController extends BaseController {
boolean isSearchId = StrUtil.isNotBlank(searchId);
boolean isSearchName = StrUtil.isNotBlank(searchName);
Comparator<YarnApplication> comparator = ComparatorUtil.longComparator(order, direction, SORT_MAP);
ImmutableList<YarnApplication> applications = yarnClusterService.jobList(Lists.immutable.ofAll(clusters))
ImmutableList<YarnApplication> applications = Lists.immutable.ofAll(clusters)
.asParallel(EXECUTOR, 1)
.flatCollect(yarnService::jobList)
.select(app -> !isFilterState || ObjectUtil.contains(filterState, app.getState()))
.select(app -> !isFilterFinalStatus || ObjectUtil.contains(filterFinalStatus, app.getFinalStatus()))
.select(app -> !isSearchId || (precise ? StrUtil.equals(app.getId(), searchId) : StrUtil.contains(app.getId(), searchId)))
@@ -90,8 +92,9 @@ public class YarnController extends BaseController {
@GetMapping("job_current")
public AmisResponse jobCurrent(@RequestParam("clusters") List<String> clusters, @RequestParam("name") String name) {
Optional<YarnApplication> currentApp = yarnClusterService.jobListEquals(Lists.immutable.ofAll(clusters), name)
Optional<YarnApplication> currentApp = Lists.immutable.ofAll(clusters)
.asParallel(EXECUTOR, 1)
.flatCollect(cluster -> yarnService.jobListEquals(cluster, name))
.select(app -> ObjectUtil.equals(app.getState(), "RUNNING"))
.toSortedList(ComparatorUtil.longComparator("startedTime", ComparatorUtil.DESC, SORT_MAP))
.getFirstOptional();
@@ -108,12 +111,12 @@ public class YarnController extends BaseController {
public AmisResponse queueList(@RequestParam("clusters") List<String> clusters, @RequestParam(value = "names", defaultValue = "") String names) {
boolean isFilterNames = StrUtil.isNotBlank(names);
ImmutableList<String> filterNames = Lists.immutable.of(names.split(","));
ImmutableList<YarnClusterVO> results = yarnClusterService.services(Lists.immutable.ofAll(clusters))
ImmutableList<YarnClusterVO> results = Lists.immutable.ofAll(clusters)
.asParallel(EXECUTOR, 1)
.collect(yarnService -> {
YarnRootQueue cluster = yarnService.cluster();
ImmutableList<YarnQueue> queues = yarnService.queueList().select(queue -> !isFilterNames || filterNames.anySatisfy(n -> StrUtil.equals(queue.getQueueName(), n)));
return new YarnClusterVO(cluster, queues);
.collect(cluster -> {
YarnRootQueue root = yarnService.cluster(cluster);
ImmutableList<YarnQueue> queues = yarnService.queueList(cluster).select(queue -> !isFilterNames || filterNames.anySatisfy(n -> StrUtil.equals(queue.getQueueName(), n)));
return new YarnClusterVO(root, queues);
})
.toList()
.toImmutable();
@@ -122,11 +125,22 @@ public class YarnController extends BaseController {
@GetMapping("queue_names")
public AmisResponse queueNames(@RequestParam("clusters") List<String> clusters) {
return responseData(MapUtil.of("queueNames", yarnClusterService.queueList(Lists.immutable.ofAll(clusters)).collect(YarnQueue::getQueueName)));
ImmutableList<String> names = Lists.immutable.ofAll(clusters)
.asParallel(EXECUTOR, 1)
.flatCollect(yarnService::queueList)
.collect(YarnQueue::getQueueName)
.toList()
.toImmutable();
return responseData(MapUtil.of("queueNames", names));
}
@GetMapping("clusters")
public AmisResponse clusters(@RequestParam("clusters") List<String> clusters) {
return responseData(MapUtil.of("cluster", yarnClusterService.cluster(Lists.immutable.ofAll(clusters))));
ImmutableList<YarnRootQueue> roots = (ImmutableList<YarnRootQueue>) Lists.immutable.ofAll(clusters)
.asParallel(EXECUTOR, 1)
.collect(yarnService::cluster)
.toList()
.toImmutable();
return responseData(MapUtil.of("cluster", clusters));
}
}