剖析azkaban解析作業(yè)流

azkaban是沒有創(chuàng)建作業(yè)流,創(chuàng)建任務,建立任務之間關系的操作的.
它本身的工作模式,在本地進行編輯.創(chuàng)建任務,創(chuàng)建作業(yè)流,建立依賴關系,然后打成zip上傳.

DirectoryFlowLoader解壓zip包,然后解析解壓出來的zip目錄.
首先需要兩個基礎類Node,Edge.
Node表示的是作業(yè)流中的節(jié)點信息,包含job文件名,properties屬性文件名,作業(yè)類型,
Edge表示的是節(jié)點之間邊信息.

loadProjectFlow

將project目錄解析成flow,properties json串,將一些無效的job,flow預先剔除掉.

無效的job:

  • type屬性沒寫的
  • 重復job(多次出現(xiàn)將會被剔除掉)

無效依賴:

  • 只身依賴(自己依賴自己)
  • 無效依賴(所依賴的job不存在)
  • 依賴于job是個重復job

代碼

public void loadProjectFlow(Project project, File baseDirectory) {
    propsList = new ArrayList<Props>();//整個工程的.properties配置列表
    flowPropsList = new ArrayList<FlowProps>();//作業(yè)流.properties配置列表
    jobPropsMap = new HashMap<String, Props>();//jobName->.job配置
    nodeMap = new HashMap<String, Node>();//jobName->Node
    flowMap = new HashMap<String, Flow>();//flowName->Flow
    errors = new HashSet<String>();
    duplicateJobs = new HashSet<String>();//重復任務名(jobName)
    nodeDependencies = new HashMap<String, Map<String, Edge>>();//jobname->sourceJobName->依賴的邊
    rootNodes = new HashSet<String>();//根節(jié)點,解釋一下,這里的根節(jié)點是flow中最后的節(jié)點
    flowDependencies = new HashMap<String, Set<String>>();//flow于flow之間的依賴關系,解決內(nèi)嵌之間依賴關系

    // Load all the props files and create the Node objects
    loadProjectFromDir(baseDirectory.getPath(), baseDirectory, null);

    jobPropertiesCheck(project);

    // Create edges and find missing dependencies
    resolveDependencies();

    // Create the flows.
    buildFlowsFromDependencies();

    // Resolve embedded flows
    resolveEmbeddedFlows();

  }

loadProjectFlow函數(shù)用于解析工作目錄,解析job,構建工作流.

第一步loadProjectFromDir(baseDirectory.getPath(), baseDirectory, null);
目的有兩個,一將配置文件解析成類,二排除文件重復job
加載.properties,.job文件,將加載的配置放入
flowPropsList,propsList.(.properties)
jobPropsMap,duplicateJobs,nodeMap.(
.job)
nodeMap存儲的是所有的節(jié)點信息

jobPropertiesCheck函數(shù),檢查job任務屬性是否合格.azkaban中會限制每個任務最大內(nèi)存和最小內(nèi)存.如果超過job的社會超過限制,就會放入error中.

  private void jobPropertiesCheck(Project project) {
    // if project is in the memory check whitelist, then we don't need to check
    // its memory settings
    if (ProjectWhitelist.isProjectWhitelisted(project.getId(),
        ProjectWhitelist.WhitelistType.MemoryCheck)) {
      return;
    }

    String maxXms = props.getString(JOB_MAX_XMS, MAX_XMS_DEFAULT);
    String maxXmx = props.getString(JOB_MAX_XMX, MAX_XMX_DEFAULT);
    long sizeMaxXms = Utils.parseMemString(maxXms);
    long sizeMaxXmx = Utils.parseMemString(maxXmx);

    for (String jobName : jobPropsMap.keySet()) {

      Props jobProps = jobPropsMap.get(jobName);
      String xms = jobProps.getString(XMS, null);
      if (xms != null && !PropsUtils.isVarialbeReplacementPattern(xms)
          && Utils.parseMemString(xms) > sizeMaxXms) {
        errors.add(String.format(
            "%s: Xms value has exceeded the allowed limit (max Xms = %s)",
            jobName, maxXms));
      }
      String xmx = jobProps.getString(XMX, null);
      if (xmx != null && !PropsUtils.isVarialbeReplacementPattern(xmx)
          && Utils.parseMemString(xmx) > sizeMaxXmx) {
        errors.add(String.format(
            "%s: Xmx value has exceeded the allowed limit (max Xmx = %s)",
            jobName, maxXmx));
      }

      // job callback properties check
      JobCallbackValidator.validate(jobName, props, jobProps, errors);
    }
  }

resolveDependencies回溯依賴關系,得到所有Edge,將無效依賴排除(依賴的job不存在,依賴的job是重復job)

buildFlowsFromDependencies利用之前解析好Nodes,Edges,回溯形成flow.

//這里所謂的根節(jié)點是末節(jié)點
private void buildFlowsFromDependencies() {
    //找出所有的依賴節(jié)點
    // Find all root nodes by finding ones without dependents.
    HashSet<String> nonRootNodes = new HashSet<String>();
    for (Map<String, Edge> edges : nodeDependencies.values()) {
      for (String sourceId : edges.keySet()) {
        nonRootNodes.add(sourceId);
      }
    }

    // Now create flows. Bad flows are marked invalid
    Set<String> visitedNodes = new HashSet<String>();
    for (Node base : nodeMap.values()) {
      // Root nodes can be discovered when parsing jobs
      if (rootNodes.contains(base.getId())
          || !nonRootNodes.contains(base.getId())) {
        rootNodes.add(base.getId());
        Flow flow = new Flow(base.getId());
        Props jobProp = jobPropsMap.get(base.getId());

        // Dedup with sets
        @SuppressWarnings("unchecked")
        List<String> successEmailList =
            jobProp.getStringList(CommonJobProperties.SUCCESS_EMAILS,
                Collections.EMPTY_LIST);
        Set<String> successEmail = new HashSet<String>();
        for (String email : successEmailList) {
          successEmail.add(email.toLowerCase());
        }

        @SuppressWarnings("unchecked")
        List<String> failureEmailList =
            jobProp.getStringList(CommonJobProperties.FAILURE_EMAILS,
                Collections.EMPTY_LIST);
        Set<String> failureEmail = new HashSet<String>();
        for (String email : failureEmailList) {
          failureEmail.add(email.toLowerCase());
        }

        @SuppressWarnings("unchecked")
        List<String> notifyEmailList =
            jobProp.getStringList(CommonJobProperties.NOTIFY_EMAILS,
                Collections.EMPTY_LIST);
        for (String email : notifyEmailList) {
          email = email.toLowerCase();
          successEmail.add(email);
          failureEmail.add(email);
        }

        flow.addFailureEmails(failureEmail);
        flow.addSuccessEmails(successEmail);

        flow.addAllFlowProperties(flowPropsList);
        constructFlow(flow, base, visitedNodes);//不斷的遞歸,直到依賴為null為止
        flow.initialize();
        flowMap.put(base.getId(), flow);
      }
    }
  }
最后編輯于
?著作權歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
  • 序言:七十年代末暖混,一起剝皮案震驚了整個濱河市健霹,隨后出現(xiàn)的幾起案子,更是在濱河造成了極大的恐慌镶柱,老刑警劉巖,帶你破解...
    沈念sama閱讀 222,627評論 6 517
  • 序言:濱河連續(xù)發(fā)生了三起死亡事件盼玄,死亡現(xiàn)場離奇詭異西疤,居然都是意外死亡,警方通過查閱死者的電腦和手機蜡秽,發(fā)現(xiàn)死者居然都...
    沈念sama閱讀 95,180評論 3 399
  • 文/潘曉璐 我一進店門,熙熙樓的掌柜王于貴愁眉苦臉地迎上來缆镣,“玉大人芽突,你說我怎么就攤上這事《埃” “怎么了寞蚌?”我有些...
    開封第一講書人閱讀 169,346評論 0 362
  • 文/不壞的土叔 我叫張陵,是天一觀的道長钠糊。 經(jīng)常有香客問我挟秤,道長,這世上最難降的妖魔是什么抄伍? 我笑而不...
    開封第一講書人閱讀 60,097評論 1 300
  • 正文 為了忘掉前任艘刚,我火速辦了婚禮,結果婚禮上截珍,老公的妹妹穿的比我還像新娘攀甚。我一直安慰自己,他們只是感情好岗喉,可當我...
    茶點故事閱讀 69,100評論 6 398
  • 文/花漫 我一把揭開白布秋度。 她就那樣靜靜地躺著,像睡著了一般钱床。 火紅的嫁衣襯著肌膚如雪荚斯。 梳的紋絲不亂的頭發(fā)上,一...
    開封第一講書人閱讀 52,696評論 1 312
  • 那天查牌,我揣著相機與錄音事期,去河邊找鬼。 笑死纸颜,一個胖子當著我的面吹牛刑赶,可吹牛的內(nèi)容都是我干的。 我是一名探鬼主播懂衩,決...
    沈念sama閱讀 41,165評論 3 422
  • 文/蒼蘭香墨 我猛地睜開眼,長吁一口氣:“原來是場噩夢啊……” “哼!你這毒婦竟也來了浊洞?” 一聲冷哼從身側(cè)響起牵敷,我...
    開封第一講書人閱讀 40,108評論 0 277
  • 序言:老撾萬榮一對情侶失蹤,失蹤者是張志新(化名)和其女友劉穎法希,沒想到半個月后枷餐,有當?shù)厝嗽跇淞掷锇l(fā)現(xiàn)了一具尸體,經(jīng)...
    沈念sama閱讀 46,646評論 1 319
  • 正文 獨居荒郊野嶺守林人離奇死亡苫亦,尸身上長有42處帶血的膿包…… 初始之章·張勛 以下內(nèi)容為張勛視角 年9月15日...
    茶點故事閱讀 38,709評論 3 342
  • 正文 我和宋清朗相戀三年毛肋,在試婚紗的時候發(fā)現(xiàn)自己被綠了。 大學時的朋友給我發(fā)了我未婚夫和他白月光在一起吃飯的照片屋剑。...
    茶點故事閱讀 40,861評論 1 353
  • 序言:一個原本活蹦亂跳的男人離奇死亡润匙,死狀恐怖,靈堂內(nèi)的尸體忽然破棺而出唉匾,到底是詐尸還是另有隱情孕讳,我是刑警寧澤,帶...
    沈念sama閱讀 36,527評論 5 351
  • 正文 年R本政府宣布巍膘,位于F島的核電站厂财,受9級特大地震影響,放射性物質(zhì)發(fā)生泄漏峡懈。R本人自食惡果不足惜璃饱,卻給世界環(huán)境...
    茶點故事閱讀 42,196評論 3 336
  • 文/蒙蒙 一、第九天 我趴在偏房一處隱蔽的房頂上張望肪康。 院中可真熱鬧荚恶,春花似錦、人聲如沸梅鹦。這莊子的主人今日做“春日...
    開封第一講書人閱讀 32,698評論 0 25
  • 文/蒼蘭香墨 我抬頭看了看天上的太陽齐唆。三九已至嗤栓,卻和暖如春,著一層夾襖步出監(jiān)牢的瞬間箍邮,已是汗流浹背茉帅。 一陣腳步聲響...
    開封第一講書人閱讀 33,804評論 1 274
  • 我被黑心中介騙來泰國打工, 沒想到剛下飛機就差點兒被人妖公主榨干…… 1. 我叫王不留锭弊,地道東北人堪澎。 一個月前我還...
    沈念sama閱讀 49,287評論 3 379
  • 正文 我出身青樓,卻偏偏與公主長得像味滞,于是被迫代替她去往敵國和親樱蛤。 傳聞我的和親對象是個殘疾皇子钮呀,可洞房花燭夜當晚...
    茶點故事閱讀 45,860評論 2 361

推薦閱讀更多精彩內(nèi)容