分布式计算任务调度系统，提供Spark、Flink等批处理任务的DAG调度和流处理任务的运行管理和状态监控

package com.meiyou.bigwhale.service; import com.meiyou.bigwhale.common.Constant; import com.meiyou.bigwhale.config.SshConfig; import com.meiyou.bigwhale.config.YarnConfig; import com.meiyou.bigwhale.data.service.AbstractMysqlPagingAndSortingQueryService; import com.meiyou.bigwhale.dto.DtoScript; import com.meiyou.bigwhale.entity.*; import com.meiyou.bigwhale.entity.auth.User; import com.meiyou.bigwhale.service.auth.UserService; import com.meiyou.bigwhale.util.WebHdfsUtils; import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.time.DateUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.domain.Page; import org.springframework.data.domain.PageImpl; import org.springframework.data.domain.PageRequest; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import javax.persistence.TypedQuery; import javax.persistence.criteria.CriteriaBuilder; import javax.persistence.criteria.CriteriaQuery; import javax.persistence.criteria.Predicate; import javax.persistence.criteria.Root; import java.text.DateFormat; import java.text.ParseException; import java.text.SimpleDateFormat; import java.util.*; import java.util.regex.Matcher; import java.util.regex.Pattern; @Service public class ScriptServiceImpl extends AbstractMysqlPagingAndSortingQueryService<Script, Integer> implements ScriptService { private static final Pattern DATE_PATTERN = Pattern.compile("(\\$\\{now(\\s*([-+])\\s*(\\d+)([dhms]))*(@([A-Za-z0-9\\s-':+.]+))*})+"); @Autowired private MonitorService monitorService; @Autowired private ScriptHistoryService scriptHistoryService; @Autowired private ClusterService clusterService; @Autowired private UserService userService; @Autowired private AgentService agentService; @Autowired private ClusterUserService clusterUserService; @Autowired private SshConfig sshConfig; @Autowired private YarnConfig yarnConfig; @Transactional(rollbackFor = Exception.class) @Override public void delete(Script entity) { //删除jar包 if (entity.isYarn()) { deleteJar(entity); } if (entity.getMonitorId() != null) { monitorService.deleteById(entity.getMonitorId()); } super.delete(entity); } @Override public Page<Script> fuzzyPage(DtoScript req) { CriteriaBuilder builder = entityManager.getCriteriaBuilder(); CriteriaQuery<Script> criteriaQuery = builder.createQuery(clazz); Root<Script> root = criteriaQuery.from(clazz); CriteriaQuery<Long> countCriteriaQuery = builder.createQuery(Long.class); Root<Script> countRoot = countCriteriaQuery.from(clazz); countCriteriaQuery.select(builder.count(countRoot)); this.predicate(builder, criteriaQuery, root, req); this.predicate(builder, countCriteriaQuery, countRoot, req); long totalCount = entityManager.createQuery(countCriteriaQuery).getSingleResult(); int pageNo = req.pageNo - 1; int pageSize = req.pageSize; TypedQuery<Script> typedQuery = entityManager.createQuery(criteriaQuery); typedQuery.setFirstResult(pageNo * pageSize); typedQuery.setMaxResults(pageSize); List<Script> entities = typedQuery.getResultList(); return new PageImpl<>(entities, new PageRequest(pageNo, pageSize), totalCount); } @Override public String validate(DtoScript req) { String msg = checkLegal(req); if (msg != null) { return msg; } //应用内存资源参数检查和补充必要参数 if (req.isYarn()) { if (yarnConfig.getAppMemoryThreshold() > 0 && !yarnConfig.getAppWhiteList().contains(req.getApp())) { try { int totalMemory = calResource(req.getType(), req.getContent()).get("totalMemory"); if (totalMemory > yarnConfig.getAppMemoryThreshold()) { //超过阀值 return "内存参数配置达到大内存应用标准【" + yarnConfig.getAppMemoryThreshold() + "MB】 ！请调小内存参数或联系管理员升级为大内存应用。"; } } catch (NumberFormatException e) { return "内存参数请向上取整"; } } appendNecessaryArgs(req); } if (req.getId() != null) { Script dbScript = findById(req.getId()); if (Constant.ScriptType.SPARK_STREAM.equals(dbScript.getType()) || Constant.ScriptType.FLINK_STREAM.equals(dbScript.getType())) { //更换集群或队列时检查应用是否正在运行 if (checkNeedKillYarnAppIfChangeClusterOrQueue(dbScript, req)) { return "更换集群或队列前请先关闭正在运行的应用"; } } if (dbScript.isYarn()) { //检查程序包是否变更 String dbJarPath = extractJarPath(dbScript.getContent()); String reqJarPath = extractJarPath(req.getContent()); if (!dbJarPath.equals(reqJarPath)) { deleteJar(dbScript); } } } if (StringUtils.isBlank(req.getUser())) { req.setUser(sshConfig.getUser()); } return null; } @Transactional(rollbackFor = Exception.class) @Override public Script update(Script entity, Monitor monitorEntity) { Monitor monitor = monitorService.save(monitorEntity); if (entity.getMonitorId() == null) { entity.setMonitorId(monitor.getId()); } return super.save(entity); } @Override public ScriptHistory generateHistory(Script script) { return generateHistory(script, null, null, null, null); } @Override public ScriptHistory generateHistory(Script script, Monitor monitor) { return generateHistory(script, monitor, null, null, null); } @Transactional(rollbackFor = Exception.class) @Override public void generateHistory(Schedule schedule, String scheduleInstanceId, String previousScheduleTopNodeId) { recursiveGenerateHistory(schedule, scheduleInstanceId, previousScheduleTopNodeId); } @Transactional(rollbackFor = Exception.class) @Override public void reGenerateHistory(Schedule schedule, String scheduleInstanceId, String previousScheduleTopNodeId) { generateHistory(schedule, scheduleInstanceId, previousScheduleTopNodeId); } @Override public String extractJarPath(String content) { String [] tokens = content.split(" "); int jarIndex = -1; for (int i = 0; i < tokens.length; i ++) { String token = tokens[i]; if (token.contains(".jar") || token.contains(".py")) { if (!"--jars".equals(tokens[i - 1]) && !"-j".equals(tokens[i - 1]) && !"--jar".equals(tokens[i - 1])) { jarIndex = i; break; } } } if (jarIndex != -1) { return tokens[jarIndex]; } return null; } @Override public void deleteJar(Script entity) { String jarPath = extractJarPath(entity.getContent()); if (jarPath != null) { //检查是否还被引用 boolean used = false; for (Script item : findByQuery("createBy=" + entity.getCreateBy() + ";id!=" + entity.getId())) { if (jarPath.equals(extractJarPath(item.getContent()))) { used = true; break; } } if (!used) { for (Cluster cluster : clusterService.findAll()) { if (jarPath.startsWith(cluster.getFsDefaultFs())) { String fs