您好,登錄后才能下訂單哦!
小編給大家分享一下Hadoop如何實現job提交,相信大部分人都還不怎么了解,因此分享這篇文章給大家參考一下,希望大家閱讀完這篇文章后大有收獲,下面讓我們一起去了解一下吧!
從如下地方開始,就要進行job的提交了
boolean isSuccess = job.waitForCompletion(true);
之后,進入Job類的waitForCompletion方法。
public boolean waitForCompletion(boolean verbose ) throws IOException, InterruptedException, ClassNotFoundException { if (state == JobState.DEFINE) { submit(); } //---- return isSuccessful(); }> > 這里輸入引用文本
之后調用Job類的submit方法,
public void submit() throws IOException, InterruptedException, ClassNotFoundException { connect(); final JobSubmitter submitter = getJobSubmitter(cluster.getFileSystem(), cluster.getClient()); status = ugi.doAs(new PrivilegedExceptionAction<JobStatus>() { public JobStatus run() throws IOException, InterruptedException, ClassNotFoundException { return submitter.submitJobInternal(Job.this, cluster); } }); }
connect方法負責初始化集群信息:
private synchronized void connect() throws IOException, InterruptedException, ClassNotFoundException { if (cluster == null) { cluster = ugi.doAs(new PrivilegedExceptionAction<Cluster>() { public Cluster run() throws IOException, InterruptedException, ClassNotFoundException { return new Cluster(getConfiguration()); } }); } }
集群信息cluster,包括什么,應該很清晰:
private ClientProtocolProvider clientProtocolProvider; private ClientProtocol client;
private UserGroupInformation ugi; private Configuration conf; private FileSystem fs = null;
private Path sysDir = null; private Path stagingAreaDir = null; private Path jobHistoryDir = null;
略微分析下, ClientProtocolProvider是客戶端協議的生產者,對應的客戶端是ClientProtocol。
ClientProtocolProvider規定了2個方法:
create
close 分別也用來創建和關閉客戶端ClientProtocol。
而,ClientProtocolProvider的具體實現類有2個。
可以看到,有兩個協議生產者,分別是yarn和local的。
那么,對應的客戶端ClientProtocol,也會有兩個。
ClientProtocol是個接口,里面規定了如下幾個方法:
那么,不同的客戶端yarn或者local,實現其中的方法即可。 因為,我們是本地Eclipse運行,直接看local即可,yarn的原理差不多,
OK,經過connect方法之后,cluster中這幾個就有啦,即使沒有的話,get的時候,也會初始化的。
之后, 使用集群的,FileSystem和client創建一個submiter。
final JobSubmitter submitter = getJobSubmitter(cluster.getFileSystem(), cluster.getClient());
然后,調用submitter 的submitJobInternal方法提交作業,OK,進入submitJobInternal方法。
JobSubmiter類的submitJobInternal方法大致過程如下:
checkSpecs(job); 檢查作業輸出路徑。
//獲得staging路徑,注意:集群cluster中有這個路徑的名稱,只不過這里需要創建路徑。 Path jobStagingArea = JobSubmissionFiles.getStagingDir(cluster, conf);
//在staging路徑下創建一個以jobid為標示的文件夾 JobID jobId = submitClient.getNewJobID(); Path submitJobDir = new Path(jobStagingArea, jobId.toString());
//job需要的一些文件和jar包之類的,都放到剛才的那個submitJobDir路徑下 copyAndConfigureFiles(job, submitJobDir);
具體的東西包括:
String files = conf.get("tmpfiles"); String libjars = conf.get("tmpjars"); String archives = conf.get("tmparchives");
//寫入job輸入的分片信息 int maps = writeSplits(job, submitJobDir);
split信息包括兩個部分。 首先調用Inputformat獲得分片的個數,具體如何獲得,后續講。 將返回的分片數組逐個遍歷并持久化到一個文件。
SplitMetaInfo[] info = writeNewSplits(conf, splits, out); 而writeNewSplits代碼主要就是寫分片信息到文件中。
之后,將split的分片信息持久化一個元數據文件。 writeJobSplitMetaInfo方法。
//將job的描述信息,寫到一個job.xml放到相應的staging目錄下的jobid目錄。 Path submitJobFile = JobSubmissionFiles.getJobConfPath(submitJobDir); writeConf(conf, submitJobFile);
FSDataOutputStream out = FileSystem.create(jtFs, jobFile, new FsPermission(JobSubmissionFiles.JOB_FILE_PERMISSION)); try { conf.writeXml(out); } finally { out.close(); }
//提交作業 status = submitClient.submitJob( jobId, submitJobDir.toString(), job.getCredentials());
OK,提交作業部分的代碼就到這,后續寫寫,app master運行的過程。
總結,提交作業的主要功能。
創建staging路徑
在staging路徑下面創建作業id的路徑
把job相關的文件拷貝到路徑下
將job的split信息序列化到文件中
將job的xml寫到路徑下
這些東西都放到hdfs,作為所有節點共享訪問的地方。之后,app master會訪問這個目錄,copy job的配置文件到本地并創建job對象,并根據split的信息,創建對應的maptaskrunable。運行。
但是,總的job信息依然在hdfs上。
以上是“Hadoop如何實現job提交”這篇文章的所有內容,感謝各位的閱讀!相信大家都有了一定的了解,希望分享的內容對大家有所幫助,如果還想學習更多知識,歡迎關注億速云行業資訊頻道!
免責聲明:本站發布的內容(圖片、視頻和文字)以原創、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。