码迷,mamicode.com
首页 > 其他好文 > 详细

Azkaban使用示例

时间:2015-01-23 11:19:02      阅读:1818      评论:0      收藏:0      [点我收藏+]

标签:azkaban   ajax   api   https   json   

1、Web端管理工作流
Azkaban提供了易操作的Web管理界面,具体操作可参考:http://azkaban.github.io/azkaban/docs/2.5/#ajax-api
值得注意的是,可以使用Azkaban提供的Web界面,定义或覆盖工作流的具体执行参数,操作界面如下:

技术分享

2.1、登录认证
命令:curl -k -X POST --data "action=login&username=azkaban&password=azkaban" https://localhost:8443
返回json格式:

<span style="white-space:pre">	</span>{
  		"status" : "success",
  		"session.id" : "c001aba5-a90f-4daf-8f11-62330d034c0a"
	}

2.2、创建一个Project
命令:curl -k -X POST --data "session.id=9089beb2-576d-47e3-b040-86dbdc7f523e&name=aaaa&description=11" https://localhost:8443/manager?action=create
返回json格式:

<span style="white-space:pre">	</span>{
  		"status":"success",
  		"path":"manager?project=aaaa",
  		"action":"redirect"
	} 

2.3、上传一个Project的zip文件
命令:curl -k -i -H "Content-Type: multipart/mixed" -X POST --form ‘session.id=e7a29776-5783-49d7-afa0-b0e688096b5e‘ --form ‘ajax=upload‘ --form ‘file=@myproject.zip;type=application/zip‘ --form ‘project=MyProject‘ https://localhost:8443/manager
注意:在zip文件的目录下执行该命令,否则无法找到zip文件
返回json格式:

<span style="white-space:pre">	</span>{
  		"error" : "Installation Failed.\nError unzipping file.",
  		"projectId" : "192",
  		"version" : "1"
	}

2.4、执行工作流
命令:curl -k --get --data ‘session.id=189b956b-f39f-421e-9a95-e3117e7543c9‘ --data ‘ajax=executeFlow‘ --data ‘project=azkaban-test-project‘ --data ‘flow=test‘ https://localhost:8443/executor
返回json格式: 

<span style="white-space:pre">	</span>{
  		message: "Execution submitted successfully with exec id 295",
  		project: "foo-demo",
  		flow: "test",
  		execid: 295
	}

3、使用Ajax API操作工作流
在程序中执行工作流的话,则需要使用Azkaban提供的ajax api了,以下是使用Java模拟https请求的代码:
AzkabanHttpsPost类

package hadoop.azkaban;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.MalformedURLException;
import java.net.URL;
import java.security.GeneralSecurityException;
import java.security.KeyStore;
import java.util.Properties;

import javax.net.ssl.HostnameVerifier;
import javax.net.ssl.HttpsURLConnection;
import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.TrustManagerFactory;

import net.sf.json.JSONObject;

public class AzkabanHttpsPost {
        static String keystorePassword;
        static String keystore;
        static String truststore;

        static{
                InputStream is=Thread.currentThread().getContextClassLoader().
                getResourceAsStream("azkaban.properties");
                Properties p=new Properties();
                try{
                        p.load(is);
                        keystorePassword = p.getProperty("PASSWORD");
                        keystore = p.getProperty("KEYSTORE");
                        truststore = p.getProperty("TRUSTSTORE");
                }catch(Exception e){
                        e.printStackTrace();
                }
        }

        /**
         * 获得KeyStore.
         * 
         * @param storePath
         *            密钥库路径
         * @param password
         *            密码
         * @return 密钥库
         * @throws Exception
         */
        public static KeyStore getKeyStore(String password, String storePath)
                        throws Exception {
                // 实例化密钥库
                KeyStore ks = KeyStore.getInstance("JKS");
                // 获得密钥库文件流
                FileInputStream is = new FileInputStream(storePath);
                // 加载密钥库
                ks.load(is, password.toCharArray());
                // 关闭密钥库文件流
                is.close();
                return ks;
        }
         /**
         * 获得SSLSocketFactory.
         * 
         * @param password
         *            密码
         * @param keyStorePath
         *            密钥库路径
         * @param trustStorePath
         *            信任库路径
         * @return SSLSocketFactory
         * @throws Exception
         */
        public static SSLContext getSSLContext() throws Exception {
                // 实例化密钥库
                KeyManagerFactory keyManagerFactory = KeyManagerFactory
                                .getInstance(KeyManagerFactory.getDefaultAlgorithm());
                // 获得密钥库
                KeyStore keyStore = getKeyStore(AzkabanHttpsPost.keystorePassword, AzkabanHttpsPost.keystore);
                // 初始化密钥工厂
                keyManagerFactory.init(keyStore, AzkabanHttpsPost.keystorePassword.toCharArray());
                // 实例化信任库
                TrustManagerFactory trustManagerFactory = TrustManagerFactory
                                .getInstance(TrustManagerFactory.getDefaultAlgorithm());
                // 获得信任库
                KeyStore trustStore = getKeyStore(AzkabanHttpsPost.keystorePassword, AzkabanHttpsPost.truststore);
                // 初始化信任库
                trustManagerFactory.init(trustStore);
                // 实例化SSL上下文
                SSLContext ctx = SSLContext.getInstance("TLS");
                // 初始化SSL上下文
                ctx.init(keyManagerFactory.getKeyManagers(),
                                trustManagerFactory.getTrustManagers(), null);
                // 获得SSLSocketFactory
                return ctx;
        }

        /**
         * 初始化HttpsURLConnection.
         * 
         * @param password
         *            密码
         * @param keyStorePath
         *            密钥库路径
         * @param trustStorePath
         *            信任库路径
         * @throws Exception
         */
        public static void initHttpsURLConnection() throws Exception {
                // 声明SSL上下文
                SSLContext sslContext = null;
                // 实例化主机名验证接口
                HostnameVerifier hnv = new MyHostnameVerifier();
                try {
                        sslContext = getSSLContext();
                } catch (GeneralSecurityException e) {
                        e.printStackTrace();
                }
                if (sslContext != null) {
                        HttpsURLConnection.setDefaultSSLSocketFactory(sslContext
                                        .getSocketFactory());
                }
                HttpsURLConnection.setDefaultHostnameVerifier(hnv);
        }
         /**
         * 发送请求.
         * 
         * @param httpsUrl
         *            请求的地址,如https://localhost:8043
         * @param xmlStr
         *            请求的数据,如action=login&username=azkaban&password=azkaban
         * @throws Exception 
         */
        public static JSONObject post(String url,String xmlStr) throws Exception {
                initHttpsURLConnection();
                JSONObject jsonObj = null;
                HttpsURLConnection urlCon = null;
                try {
                        urlCon = (HttpsURLConnection) (new URL(url)).openConnection();
                        urlCon.setDoInput(true);
                        urlCon.setDoOutput(true);
                        urlCon.setRequestMethod("POST");
                        // 如下设置后,azkaban才能识别出是以ajax的方式访问,从而返回json格式的操作信息
                        urlCon.setRequestProperty("Content-Type",
                                        "application/x-www-form-urlencoded");
                        urlCon.setRequestProperty("X-Requested-With", "XMLHttpRequest");
                        urlCon.setUseCaches(true);
                        // 设置为gbk可以解决服务器接收时读取的数据中文乱码问题
                        urlCon.getOutputStream().write(xmlStr.getBytes("gbk"));
                        urlCon.getOutputStream().flush();
                        urlCon.getOutputStream().close();
                        BufferedReader in = new BufferedReader(new InputStreamReader(
                                        urlCon.getInputStream()));
                        String line="";
                        String temp;
                        while ((temp = in.readLine()) != null) {
                                line = line + temp;
                        }
                        jsonObj = JSONObject.fromObject(line);
                } catch (MalformedURLException e) {
                        e.printStackTrace();
                } catch (IOException e) {
                        e.printStackTrace();
                } catch (Exception e) {
                        e.printStackTrace();
                }
                return jsonObj;
        }

MyHostnameVerifier类

package hadoop.azkaban;
import javax.net.ssl.HostnameVerifier;  
import javax.net.ssl.SSLSession;  
  
/** 
 * 实现用于主机名验证的基接口。  
 * 在握手期间,如果 URL 的主机名和服务器的标识主机名不匹配,则验证机制可以回调此接口的实现程序来确定是否应该允许此连接。 
 */  
public class MyHostnameVerifier implements HostnameVerifier {  
    @Override  
    public boolean verify(String hostname, SSLSession session) {  
        if("localhost".equals(hostname)){  
            return true;  
        } else {  
            return false;  
        }  
    }  
}  

以下是调用https,操作Azkaban的示例:

package hadoop.azkaban;

import java.io.InputStream;
import java.util.Properties;

import net.sf.json.JSONObject;

/**
 * 
 * @author hu
 *
 */
public class AzkabanOperator {
        public static String url;
        public static String azkabanUser;
        public static String azkabanPassword;
        public static String GDI_Project;
        public static String GDI_Workflow;
        static {
                InputStream is = Thread.currentThread().getContextClassLoader()
                                .getResourceAsStream("azkaban.properties");
                Properties p = new Properties();
                try {
                        p.load(is);
                        url = p.getProperty("URL");
                        azkabanUser = p.getProperty("AZKABANUSER");
                        azkabanPassword = p.getProperty("AZKABANPASSWORD");
                        GDI_Project = p.getProperty("GDI_Project");
                        GDI_Workflow = p.getProperty("GDI_Workflow");
                } catch (Exception e) {
                        e.printStackTrace();
                }
        }

        public JSONObject login() throws Exception {
                JSONObject result = null;
                String queryStr = "action=login&username=" + azkabanUser + "&password="
                                + azkabanPassword;
                result = AzkabanHttpsPost.post(url, queryStr);
                return result;
        }
         public JSONObject executeGDIFlow(String sessionID, String project,
                        String flow, String cwParams, String smParams, String gdiParams)
                        throws Exception {
                JSONObject result = null;
                String executeStr = "session.id=" + sessionID
                                + "&ajax=executeFlow&project=" + project + "&flow=" + flow
                                + "&flowOverride[cw_params]=" + cwParams
                                + "&flowOverride[sm_params]=" + smParams
                                + "&flowOverride[gdi_params]=" + gdiParams;
                String executeUrl = url + "/executor";
                result = AzkabanHttpsPost.post(executeUrl, executeStr);
                return result;
        }

        public JSONObject fetchFlow(String sessionID, String execID)
                        throws Exception {
                JSONObject result = null;
                String executeStr = "session.id=" + sessionID
                                + "&ajax=fetchexecflow&execid=" + execID;
                String executeUrl = url + "/executor";
                result = AzkabanHttpsPost.post(executeUrl, executeStr);
                return result;
        }

Azkaban使用示例

标签:azkaban   ajax   api   https   json   

原文地址:http://blog.csdn.net/until_v/article/details/43052193

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!