這篇文章主要講解了“Hadoop Job提交相關(guān)知識(shí)點(diǎn)分析”，文中的講解內(nèi)容簡(jiǎn)單清晰，易于學(xué)習(xí)與理解，下面請(qǐng)大家跟著小編的思路慢慢深入，一起來(lái)研究和學(xué)習(xí)“Hadoop Job提交相關(guān)知識(shí)點(diǎn)分析”吧！

創(chuàng)新互聯(lián)是一家專(zhuān)注于網(wǎng)站制作、做網(wǎng)站與策劃設(shè)計(jì),仁化網(wǎng)站建設(shè)哪家好?創(chuàng)新互聯(lián)做網(wǎng)站,專(zhuān)注于網(wǎng)站建設(shè)10余年,網(wǎng)設(shè)計(jì)領(lǐng)域的專(zhuān)業(yè)建站公司;建站業(yè)務(wù)涵蓋:仁化等地區(qū)。仁化做網(wǎng)站價(jià)格咨詢(xún):18980820575

Configuration類(lèi)是用來(lái)訪問(wèn)hadoop的配置參數(shù)的。

Configuration類(lèi)首先會(huì)通過(guò)靜態(tài)代碼段加載hadoop的配置文件core-default.xml和和core-site.xml，相關(guān)代碼如下：

<span >static{ 
    //print deprecation warning if hadoop-site.xml is found in classpath 
    ClassLoader cL = Thread.currentThread().getContextClassLoader(); 
    if (cL == null) { 
      cL = Configuration.class.getClassLoader(); 
    } 
    if(cL.getResource("hadoop-site.xml")!=null) { 
      LOG.warn("DEPRECATED: hadoop-site.xml found in the classpath. " + 
          "Usage of hadoop-site.xml is deprecated. Instead use core-site.xml, " 
          + "mapred-site.xml and hdfs-site.xml to override properties of " + 
          "core-default.xml, mapred-default.xml and hdfs-default.xml " + 
          "respectively"); 
    } 
    addDefaultResource("core-default.xml"); 
    addDefaultResource("core-site.xml"); 
  } </span>

defaultResources是一個(gè)ArrayList，用來(lái)保存默認(rèn)的配置文件路徑。如果一個(gè)默認(rèn)的配置文件路徑不在defaultResource里面，就添加進(jìn)去，這個(gè)邏輯是在

addDefaultResource方法中實(shí)現(xiàn)的。

properties是一個(gè)Properties對(duì)象，保存從配置文件中解析出來(lái)的配置屬性，如果多個(gè)配置文件有相同的key，后者會(huì)覆蓋前者的值。

JobConf類(lèi)用來(lái)配置Map/Reduce作業(yè)信息的，繼承自Configuration類(lèi)。

JobConf類(lèi)首先會(huì)通過(guò)靜態(tài)代碼段加載mapred-default.xml和mapred-site.xml配置屬性文件。

DEFAULT_MAPRED_TASK_JAVA_OPTS=“-Xmx200m”，默認(rèn)情況下Map/Reduce任務(wù)的JAVA命令行選項(xiàng)指定的JAVA虛擬機(jī)最大內(nèi)存是200M。

JobClient類(lèi)是用戶(hù)與JobTracker交互的主要接口，通過(guò)它可以提交jobs，追蹤job的進(jìn)度，訪問(wèn)task組件的日志，查詢(xún)集群的狀態(tài)信息等。

提交job是通過(guò)runJob方法實(shí)現(xiàn)的，相關(guān)代碼如下：

<span >public static RunningJob runJob(JobConf job) throws IOException {
    JobClient jc = new JobClient(job);
    RunningJob rj = jc.submitJob(job);
    try {
      if (!jc.monitorAndPrintJob(job, rj)) {
        LOG.info("Job Failed: " + rj.getFailureInfo());
        throw new IOException("Job failed!");
      }
    } catch (InterruptedException ie) {
      Thread.currentThread().interrupt();
    }
    return rj;
  }</span>

首先創(chuàng)建一個(gè)JobClient對(duì)象，此對(duì)象在構(gòu)造函數(shù)中會(huì)根據(jù)JobConf對(duì)象去連接JobTracker。

 JobClient與JobTracker通信是通過(guò)jobSubmitClient操作的，jobSubmitClient是JobSubmissionProtocol類(lèi)型的動(dòng)態(tài)代理類(lèi)，是通過(guò)如下方法產(chǎn)生的：

<span >  private static JobSubmissionProtocol createRPCProxy(InetSocketAddress addr,
      Configuration conf) throws IOException {
    return (JobSubmissionProtocol) RPC.getProxy(JobSubmissionProtocol.class,
        JobSubmissionProtocol.versionID, addr, 
        UserGroupInformation.getCurrentUser(), conf,
        NetUtils.getSocketFactory(conf, JobSubmissionProtocol.class));
  }</span>

getProxy方法的關(guān)鍵是Invoker類(lèi)，Invoker類(lèi)實(shí)現(xiàn)了InvocationHandler接口，主要有兩個(gè)成員變量，remoteId是Client.ConnectionId類(lèi)型，保存連接地址和用戶(hù)的ticket，客戶(hù)端連接服務(wù)器由<remoteAddress，protocol，ticket>唯一標(biāo)識(shí)。從這里我們也可以看到一些配置屬性值，默認(rèn)的rpcTimeout是0，

ipc.client.connection.maxidletime客戶(hù)端連接的最大空閑時(shí)間是10s，

ipc.client.connect.max.retries客戶(hù)端同服務(wù)器建立連接時(shí)的最大重試次數(shù)是10，

ipc.client.tcpnodelay是否開(kāi)啟Nagle算法（對(duì)TCP/IP進(jìn)行擁塞控制），如果開(kāi)啟，會(huì)減少延遲，但是會(huì)增加小數(shù)據(jù)報(bào)，默認(rèn)是false。client是Client類(lèi)，用于IPC通信。client會(huì)通過(guò)ClientCache類(lèi)來(lái)緩存，如果緩存中沒(méi)有，會(huì)新建一個(gè)Client，否則原client計(jì)數(shù)加1。Invoker類(lèi)主要的方法是invoke方法，invoke方法的功能是調(diào)用client的方法然后返回結(jié)果。動(dòng)態(tài)代理類(lèi)代理的對(duì)象是Client對(duì)象。

submitJobInternal方法是真正用來(lái)提交job的，具體步驟如下：

1、初始化staging目錄，staging目錄根目錄是由mapreduce.jobtracker.staging.root.dir配置的，默認(rèn)是/tmp/hadoop/mapred/staging，具體到某個(gè)用戶(hù)的staging目錄是$ROOT/userName/.staging。

2、從JobTracker那里取得新的job id，job id從1開(kāi)始遞增。

3、獲得提交job的目錄submitJobDir=用戶(hù)的staging目錄/jobid，并且將這個(gè)目錄設(shè)置成mapreduce.job.dir的值。

 4、copyAndConfigureFiles拷貝和初始化文件，首先從配置屬性mapred.submit.replication取得replication值，默認(rèn)為10。然后判斷submitJobDir目錄是否存在，如果存在拋異常；否則創(chuàng)建submitJobDir目錄；取得job的分布式緩存文件路徑=submitJobDir/files；取得job的分布式緩存存檔路徑=submitJobDir/archives；取得job的分布式緩存libjars路徑=submitJobDir/libjars；如果命令行參數(shù)有tmpfiles，則將這些文件拷貝到分布式緩存文件路徑下，同時(shí)將這個(gè)路徑加入到分布式緩存中；如果命令行參數(shù)有tmpjars，則將這些文件拷貝到分布式緩存libjars路徑下，同時(shí)將這個(gè)路徑加入到分布式緩存中；如果命令行參數(shù)有tmparchives，則將這些文件拷貝到分布式緩存存檔路徑下，同時(shí)將這個(gè)路徑加入到分布式緩存中；根據(jù)mapred.jar屬性取得jar包的路徑，如果沒(méi)有指定job的名字，那么將使用jar包的名字作為job名字；取得job jar的存儲(chǔ)路徑=submitJobDir/job.jar；將用戶(hù)指定的jar包拷貝到j(luò)ob jar的存儲(chǔ)路徑；設(shè)置工作目錄，默認(rèn)是配置屬性mapred.working.dir指定的值。

5、取得job配置文件的路徑submitJobFile=submitJobDir/job.xml；設(shè)置

mapreduce.job.submithostaddress為本機(jī)ip地址，設(shè)置

mapreduce.job.submithost為本機(jī)主機(jī)名。

6、為job創(chuàng)建輸入分區(qū)，這是由writeSplits方法完成的。以old api為例，首先調(diào)用InputFormat的getSplits方法得到一個(gè)InputSplit分區(qū)數(shù)組，F(xiàn)ileInputFormat類(lèi)的getSplits方法實(shí)現(xiàn)過(guò)程如下：

通過(guò)listStatus方法取得輸入文件路徑列表，過(guò)濾掉_和.開(kāi)頭的路徑以及根據(jù)設(shè)置的mapred.input.pathFilter.class過(guò)濾；

在JobConf中設(shè)置mapreduce.input.num.files為輸入文件數(shù)；

計(jì)算出所有輸入文件的總大小totalSize，目標(biāo)分區(qū)大小goalSize=totalSize/numSplits(由mapred.map.tasks配置，默認(rèn)為1)，最小分區(qū)大小minSize=mapred.min.split.size配置和1之間的較大值，對(duì)于每一個(gè)輸入文件，如果這個(gè)文件的長(zhǎng)度不等于0并且是可切分的，計(jì)算分區(qū)大小splitSize=Math.max(minSize,Math.min(goalSize,blockSize))，blockSize為HDFS存儲(chǔ)文件的塊大小，對(duì)于每一個(gè)分區(qū)大小，計(jì)算對(duì)其貢獻(xiàn)最大的主機(jī)數(shù)組（根據(jù)機(jī)架以及塊的字節(jié)大小確定），然后將這個(gè)分區(qū)加入到分區(qū)列表；然后根據(jù)分區(qū)大長(zhǎng)度從大到小對(duì)分區(qū)列表進(jìn)行排序；然后將分區(qū)列表寫(xiě)入到分區(qū)文件，分區(qū)文件名=submitJobDir/job.split，分區(qū)文件的存儲(chǔ)格式：SPL字節(jié)信息，分區(qū)版本號(hào)，{InputSplit類(lèi)名，InputSplit類(lèi)信息}+；SplitMetaInfo數(shù)組記錄每個(gè)分區(qū)信息在文件中的偏移，主機(jī)信息和長(zhǎng)度；將分區(qū)Meta信息SplitMetaInfo數(shù)組寫(xiě)入到文件submitJobDir/job.splitmetainfo。

7、JobConf設(shè)置mapred.map.tasks為分區(qū)數(shù)。

8、根據(jù)mapred.job.queue.name獲得job提交的隊(duì)列的名字，默認(rèn)是default，然后根據(jù)這個(gè)隊(duì)列名字獲得訪問(wèn)控制列表。

9、將重新配置過(guò)的JobConf寫(xiě)入到submitJobDir/job.xml文件。

10、將jobid，submitJobDir信息傳給JobTracker正式提交job，并通過(guò)NetworkedJob對(duì)象跟蹤job的狀態(tài)。

monitorAndPrintJob方法監(jiān)控job的運(yùn)行并且實(shí)時(shí)打印job的狀態(tài)。

感謝各位的閱讀，以上就是“Hadoop Job提交相關(guān)知識(shí)點(diǎn)分析”的內(nèi)容了，經(jīng)過(guò)本文的學(xué)習(xí)后，相信大家對(duì)Hadoop Job提交相關(guān)知識(shí)點(diǎn)分析這一問(wèn)題有了更深刻的體會(huì)，具體使用情況還需要大家實(shí)踐驗(yàn)證。這里是創(chuàng)新互聯(lián)，小編將為大家推送更多相關(guān)知識(shí)點(diǎn)的文章，歡迎關(guān)注！

本文題目：HadoopJob提交相關(guān)知識(shí)點(diǎn)分析
標(biāo)題鏈接：http://m.rwnh.cn/article18/jcjcdp.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供企業(yè)網(wǎng)站制作、云服務(wù)器、標(biāo)簽優(yōu)化、App設(shè)計(jì)、網(wǎng)站排名、網(wǎng)站營(yíng)銷(xiāo)

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶(hù)投稿、用戶(hù)轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來(lái)源： 創(chuàng)新互聯(lián)