使用java在aliyun/aws创建E-MapReduce (emr)集群

本文主要是介绍使用java在aliyun/aws创建E-MapReduce (emr)集群，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

背景

在上个公司，我的 hera 任务调度系统是运行在本地 cdh 机器上的，并没有使用 aws/aliyun 提供的 emr 服务。所以为了使 hera 能够兼容 emr，就需要使用 java 创建 emr 集群.

功能点

既然要创建集群，肯定也要有等待集群创建完成、销毁集群等操作。

所以功能点大概有

• 判断集群是否已经创建过
• 创建集群
• 等待集群创建完成
• 集群销毁
• 获得集群的登录脚本
• 集群弹性伸缩
• 等等

开撸

emr接口

为了增加程序的可维护性和扩展性，我提供了一个Emr接口类

/*** desc:** @author scx* @create 2019/04/01*/
public interface Emr {/*** 添加任务接口*/void addJob();/*** 移除任务*/void removeJob();/*** 获得登录命令* @return  登录命令*/String getLogin(String user);}

主要有三个方法，其中 addJob 和 removeJob 的作用是为了后面进行判断集群是否有任务执行来进行关闭集群的操作。而 getLogin 方法是为了获得集群的 ssh 登录命令

emr抽象类

我希望在抽象类中完成所有的同步操作、集群关闭监听事件、等待集群创建完成 等等公共操作，而子类只关注一些简单实现即可。具体请看源码

/*** desc: emr集群 抽象类** @author scx* @create 2019/04/01*/
public abstract class AbstractEmr implements Emr {/*** 缓存的集群IP*/protected volatile String cacheIp = null;/*** 任务数*/private AtomicInteger taskRunning;/*** 任务计数器*/private AtomicLong taskNum;/*** 缓存的集群Id*/private String cacheClusterId;/*** 关闭集群的调度池*/private ScheduledExecutorService pool;/*** 上次的任务数*/private long cacheTaskNum;/*** 集群是否已经关闭字段*/private volatile boolean clusterTerminate = true;/*** check 集群是否需要关闭返回的future*/private ScheduledFuture<?> clusterWatchFuture;/*** 当前应用的环境 设置不同的参数*/private String env = HeraGlobalEnvironment.getEnv() == null ? "daily" : HeraGlobalEnvironment.getEnv();/*** emr集群的前缀*/protected final String clusterPrefix = "hera-schedule-" + env + "-";/*** 自动扩展冷却时间 单位：秒*/private int coolDown = 300;/*** 扩展百分比*/private int scalePercent = 10;/*** 最少实例数*/private int minCapacity = 1;/*** 最大实例数*/private int maxCapacity = 10;public int getCoolDown() {return coolDown;}public int getScalePercent() {return scalePercent;}public int getMinCapacity() {return minCapacity;}public int getMaxCapacity() {return maxCapacity;}protected String getClusterName() {return clusterPrefix + ActionUtil.getCurrDate();}protected String getEnv() {return env;}/*** 创建集群的方法*/protected void createCluster() {//抽象方法留给子类实现init();//dubbo-check 同步，避免多次创建if (clusterTerminate) {synchronized (this) {if (clusterTerminate) {//启动时可能已经有创建好的集群 首先判断if (notAlive()) {//如果没有创建好的集群，发送创建集群的请求，抽象方法具体留给子类实现cacheClusterId = sendCreateRequest();}//参数初始化clusterTerminate = false;taskRunning = new AtomicInteger(0);taskNum = new AtomicLong(0);//等待集群创建完成waitClusterCompletion();//集群关闭的监视器submitClusterWatch();MonitorLog.info("集群创建完成,可以执行任务了.集群ID为：" + cacheClusterId + ".集群IP为:" + getIp());}}} else {//此时虽然集群被判断为活着，但是也有可能被人为关闭，recheckif (notAlive()) {destroyCluster();createCluster();}}}@Overridepublic void addJob() {//每次调用addJob方法都要执行一下createCluster方法createCluster();//正在执行的任务数+1taskRunning.incrementAndGet();//总任务数计数taskNum.incrementAndGet();}@Overridepublic void removeJob() {//正在执行的任务数-1if (taskRunning != null) {taskRunning.decrementAndGet();}}/*** 判断是否有创建好的集群* @return  结果*/private boolean notAlive() {//getAliveId 抽象方法留给具体子类实现，返回已经创建好的集群idreturn StringUtils.isBlank(cacheClusterId = getAliveId());}/*** 获得emr集群的登录ip/域名* @return  ip/域名*/public String getIp() {//同样dubbo-check 避免多次判断if (cacheIp == null) {synchronized (this) {if (cacheIp == null) {createCluster();// Amazon返回的是域名 ，aliyun目前只返回了master的内网ip,可以自己自定义 getMasterIp 的实现cacheIp = getMasterIp(cacheClusterId);if (StringUtils.isBlank(cacheIp)) {throw new NullPointerException("cacheIp can not be null");}}}}return cacheIp;}/*** 循环检测 ，等待集群创建完成*/private void waitClusterCompletion() {long start = System.currentTimeMillis();long sleepTime = 15 * 1000 * 1000000L;// 必须同步 不能异步while (!isCompletion(cacheClusterId)) {LockSupport.parkNanos(sleepTime);}MonitorLog.info("创建集群:" + cacheClusterId + "耗时:" + (System.currentTimeMillis() - start) + "ms");}/*** createCluster 方法已经同步过* 关于为什么为这里使用的11分钟:我的hera任务调度系统在任务失败后有任务重试，* 重试的间隔时间为10分钟，由于创建集群的时间过长，* 避免任务在重试期间集群被关闭，然后又重新创建，* 所以选择了略大于10的值11。具体时间，大家可以自己衡量*/private void submitClusterWatch() {if (clusterWatchFuture == null) {if (pool == null) {pool = new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("cluster-destroy-watch", true));}//11分钟前的任务数cacheTaskNum = taskNum.get();clusterWatchFuture = pool.scheduleWithFixedDelay(() -> {MonitorLog.info("正在emr集群运行的任务个数:{},十分钟前运行的总任务个数:{},现在运行的总任务个数:{}", taskRunning.get(), cacheTaskNum, taskNum.get());//如果正在执行的任务数为0并且上次记录的任务总数与当前的任务总数一致，考虑关闭集群if (taskRunning.get() == 0 && cacheTaskNum == taskNum.get()) {terminateCluster(cacheClusterId);MonitorLog.info("集群:" + cacheClusterId + "关闭成功,执行任务数为:" + taskNum.get());//关闭调度clusterWatchFuture.cancel(true);} else {//重新设置任务数cacheTaskNum = taskNum.get();}}, 11, 11, TimeUnit.MINUTES);}}/*** 初始化client操作*/protected abstract void init();/*** 关闭集群操作** @param clusterId clusterId*/protected abstract void terminateCluster(String clusterId);/*** 关闭client*/protected abstract void shutdown();/*** 发送创建集群的请求** @return 返回clusterId*/protected abstract String sendCreateRequest();/*** 判断以clusterPrefix开头的机器是否有存活** @return StringUtils.isBlank() 表示无存活*/protected abstract String getAliveId();/*** 检测集群是否创建完成** @param clusterId clusterId* @return*/protected abstract boolean isCompletion(String clusterId);/*** 获得master的ip** @param clusterId clusterId* @return*/protected abstract String getMasterIp(String clusterId);/*** 销毁集群，以及做一些后置操作*/protected synchronized void destroyCluster() {if (!clusterTerminate) {clusterTerminate = true;cacheIp = null;cacheClusterId = null;clusterWatchFuture = null;pool.shutdown();pool = null;shutdown();}}
}

阿里云集群的创建

这篇关于使用java在aliyun/aws创建E-MapReduce (emr)集群的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---