Azkaban使用示例

<span style="white-space:pre"> </span>{ "status" : "success", "session.id" : "c001aba5-a90f⑷daf⑻f11⑹2330d034c0a" }

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

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

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

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; }

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; } } }

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; }