您好,登錄后才能下訂單哦!
通過 job.waitForCompletion(true)完成job的提交以及運行,下面從這個方法入手分析源碼。
//-----------------job.java
public boolean waitForCompletion(boolean verbose) throws IOException, InterruptedException, ClassNotFoundException {
//如果job的狀態(tài)為未運行,則提交任務
if (this.state == Job.JobState.DEFINE) {
this.submit();
}
if (verbose) {
//監(jiān)控并打印運行信息
this.monitorAndPrintJob();
} else {
int completionPollIntervalMillis = getCompletionPollInterval(this.cluster.getConf());
while(!this.isComplete()) {
try {
Thread.sleep((long)completionPollIntervalMillis);
} catch (InterruptedException var4) {
}
}
}
return this.isSuccessful();
}
//-----------------job.java
public void submit() throws IOException, InterruptedException, ClassNotFoundException {
//確定job狀態(tài)為未運行
this.ensureState(Job.JobState.DEFINE);
//使用新api
this.setUseNewAPI();
//主要就是初始化cluster對象中的client,用于和集群連接通信。分為yarn client和local client
this.connect();
//通過cluster對象獲取job提交器,將存儲job信息的文件系統(tǒng)以及client作為參數(shù)
final JobSubmitter submitter = this.getJobSubmitter(this.cluster.getFileSystem(), this.cluster.getClient());
//提交job,并運行
this.status = (JobStatus)this.ugi.doAs(new PrivilegedExceptionAction<JobStatus>() {
public JobStatus run() throws IOException, InterruptedException, ClassNotFoundException {
//這里是提交job,運行,返回狀態(tài)
return submitter.submitJobInternal(Job.this, Job.this.cluster);
}
});
this.state = Job.JobState.RUNNING;
LOG.info("The url to track the job: " + this.getTrackingURL());
}
上面這里涉及到三個重要過程方法:
this.connect() 主要初始化了提交job的client
this.getJobSubmitter() 給job封裝了很多api
submitter.submitJobInternal(Job.this, Job.this.cluster) 提交job,并運行
下面詳細看看這三個方法具體做了啥
//-----------------job.java
private synchronized void connect() throws IOException, InterruptedException, ClassNotFoundException {
//創(chuàng)建cluster連接對象,用于連接集群,提供了很多api
if (this.cluster == null) {
this.cluster = (Cluster)this.ugi.doAs(new PrivilegedExceptionAction<Cluster>() {
public Cluster run() throws IOException, InterruptedException, ClassNotFoundException {
return new Cluster(Job.this.getConfiguration());
}
});
}
}
這代碼最重要的就是創(chuàng)建了一個 Cluster對象,下面看看這個類的構(gòu)造方法。
//----------------------------Cluster.java
public Cluster(Configuration conf) throws IOException {
this((InetSocketAddress)null, conf);
}
public Cluster(InetSocketAddress jobTrackAddr, Configuration conf) throws IOException {
this.fs = null;
this.sysDir = null;
//job工作目錄
this.stagingAreaDir = null;
this.jobHistoryDir = null;
//客戶端和server通信協(xié)議提供者
this.providerList = null;
//將job的配置conf保存
this.conf = conf;
//獲取當前用戶
this.ugi = UserGroupInformation.getCurrentUser();
//對job提交器client進行初始化
this.initialize(jobTrackAddr, conf);
}
//這里就是初始化client的方法了,主要就是獲得 this.client
private void initialize(InetSocketAddress jobTrackAddr, Configuration conf) throws IOException {
this.initProviderList();
Iterator i$ = this.providerList.iterator();
while(i$.hasNext()) {
/*
provider這里也有分 YarnClientProtocolProvider 以及LocalClientProtocolProvider
即本地和yarn兩種provider
*/
ClientProtocolProvider provider = (ClientProtocolProvider)i$.next();
LOG.debug("Trying ClientProtocolProvider : " + provider.getClass().getName());
ClientProtocol clientProtocol = null;
try {
/*判斷jobTrackAddr是否為空,也就是以遠程集群還是本地的方式運行job.
遠程集群的話,就創(chuàng)建yarn 提交器,:YARNRunner,通過YarnClientProtocolProvider創(chuàng)建
本地的話,就創(chuàng)建本地local 提交器:LocalRunner,通過 LocalClientProtocolProvider創(chuàng)建
主要是根據(jù) mapreduce.framework.name 在conf中的值是local還是yarn來創(chuàng)建對應的runner
*/
if (jobTrackAddr == null) {
clientProtocol = provider.create(conf);
} else {
clientProtocol = provider.create(jobTrackAddr, conf);
}
if (clientProtocol != null) {
this.clientProtocolProvider = provider;
//可以看到這里client就是上面通過provider創(chuàng)建的
this.client = clientProtocol;
LOG.debug("Picked " + provider.getClass().getName() + " as the ClientProtocolProvider");
//只要成功創(chuàng)建了client 和 provider就退出
break;
}
LOG.debug("Cannot pick " + provider.getClass().getName() + " as the ClientProtocolProvider - returned null protocol");
} catch (Exception var7) {
LOG.info("Failed to use " + provider.getClass().getName() + " due to error: ", var7);
}
}
if (null == this.clientProtocolProvider || null == this.client) {
throw new IOException("Cannot initialize Cluster. Please check your configuration for mapreduce.framework.name and the correspond server addresses.");
}
}
可以看到Cluster對象主要就是初始化了 clientProtocolProvider 以及 client 兩個對象。
也就是provider和client,client是通過provider.create創(chuàng)建的。
下面可以看看ClientProtocolProvider和 ClientProtocol這兩個類。這兩個類都是抽象類,那么看他們對應有哪些實現(xiàn)子類。
ClientProtocolProvider:
YarnClientProtocolProvider
LocalClientProtocolProvider
ClientProtocol:
YARNRunner
LocalJobRunner
可以看看YarnClientProtocolProvider 以及LocalClientProtocolProvider的create方法
public class LocalClientProtocolProvider extends ClientProtocolProvider {
.........
public ClientProtocol create(Configuration conf) throws IOException {
String framework = conf.get("mapreduce.framework.name", "local");
if (!"local".equals(framework)) {
return null;
} else {
conf.setInt("mapreduce.job.maps", 1);
//創(chuàng)建LocalJobRunner
return new LocalJobRunner(conf);
}
}
.....................
}
//===============================================================
public class YarnClientProtocolProvider extends ClientProtocolProvider {
...................................
public ClientProtocol create(Configuration conf) throws IOException {
//創(chuàng)建 YARNRunner
return "yarn".equals(conf.get("mapreduce.framework.name")) ? new YARNRunner(conf) : null;
}
...........................
}
總的來說,就是provider分為YarnClientProtocolProvider 以及LocalClientProtocolProvider,分別用于創(chuàng)建client中的 YARNRunner 和 LocalJobRunner。表示job運行方式有本地和yarn兩種。
至此,this.client以及this.provider這兩個在Cluster對象中的對象初始化完成。
//-------------------job.java
public JobSubmitter getJobSubmitter(FileSystem fs, ClientProtocol submitClient) throws IOException {
return new JobSubmitter(fs, submitClient);
}
創(chuàng)建個 JobSubmitter對象,看看構(gòu)造方法
//------------------JobSubmitter.java
JobSubmitter(FileSystem submitFs, ClientProtocol submitClient) throws IOException {
this.submitClient = submitClient;
this.jtFs = submitFs;
}
看起來,沒啥特別, 就是把文件系統(tǒng)fs以及 上面cluster中初始化的client保存起來。但是其實這個類中有很多方法后面會調(diào)用。后面講
這個方法是整個job提交過程中的核心,要注意看
//------------------JobSubmitter.java
JobStatus submitJobInternal(Job job, Cluster cluster) throws ClassNotFoundException, InterruptedException, IOException {
//檢查配置的輸出是否已存在,已存在會拋出異常
this.checkSpecs(job);
Configuration conf = job.getConfiguration();
addMRFrameworkToDistributedCache(conf);
//獲取所有job工作總目錄
Path jobStagingArea = JobSubmissionFiles.getStagingDir(cluster, conf);
//獲取ip地址對象
InetAddress ip = InetAddress.getLocalHost();
//設(shè)置提交job的主機名和ip
if (ip != null) {
this.submitHostAddress = ip.getHostAddress();
this.submitHostName = ip.getHostName();
conf.set("mapreduce.job.submithostname", this.submitHostName);
conf.set("mapreduce.job.submithostaddress", this.submitHostAddress);
}
//通過client向集群申請運行job,獲取到對應的jobid.這個submitclient是前面cluster初始化完成的
JobID jobId = this.submitClient.getNewJobID();
job.setJobID(jobId);
//創(chuàng)建存儲job相關(guān)資源數(shù)據(jù)的目錄對象.存儲job配置文件、切片信息文件、程序jar包等
Path submitJobDir = new Path(jobStagingArea, jobId.toString());
JobStatus status = null;
JobStatus var24;
try {
conf.set("mapreduce.job.user.name", UserGroupInformation.getCurrentUser().getShortUserName());
conf.set("hadoop.http.filter.initializers", "org.apache.hadoop.yarn.server.webproxy.amfilter.AmFilterInitializer");
conf.set("mapreduce.job.dir", submitJobDir.toString());
LOG.debug("Configuring job " + jobId + " with " + submitJobDir + " as the submit dir");
//獲取訪問namenode中特定目錄授權(quán)
TokenCache.obtainTokensForNamenodes(job.getCredentials(), new Path[]{submitJobDir}, conf);
this.populateTokenCache(conf, job.getCredentials());
//驗證token相關(guān)
if (TokenCache.getShuffleSecretKey(job.getCredentials()) == null) {
KeyGenerator keyGen;
try {
keyGen = KeyGenerator.getInstance("HmacSHA1");
keyGen.init(64);
} catch (NoSuchAlgorithmException var19) {
throw new IOException("Error generating shuffle secret key", var19);
}
SecretKey shuffleKey = keyGen.generateKey();
TokenCache.setShuffleSecretKey(shuffleKey.getEncoded(), job.getCredentials());
}
if (CryptoUtils.isEncryptedSpillEnabled(conf)) {
conf.setInt("mapreduce.am.max-attempts", 1);
LOG.warn("Max job attempts set to 1 since encrypted intermediatedata spill is enabled");
}
//復制job的臨時文件,以及運行的jar包到submitJobDir下
this.copyAndConfigureFiles(job, submitJobDir);
//獲取存儲job配置信息文件路徑,一般命名為:submitJobDir/job.xml
Path submitJobFile = JobSubmissionFiles.getJobConfPath(submitJobDir);
LOG.debug("Creating splits at " + this.jtFs.makeQualified(submitJobDir));
//將切片信息存儲到submitJobDir下,并返回切片數(shù)目。會調(diào)用 InputFormat.getSplits()來獲取規(guī)劃的切片信息
//切片信息會寫入到 submitJobDir/job.split,切片信息條目的元信息寫入到 submitJobDir/job.splitmetainfo
int maps = this.writeSplits(job, submitJobDir);
conf.setInt("mapreduce.job.maps", maps);
LOG.info("number of splits:" + maps);
//傳輸隊列名稱
String queue = conf.get("mapreduce.job.queuename", "default");
//submitClient其實就是cluster的client
AccessControlList acl = this.submitClient.getQueueAdmins(queue);
conf.set(QueueManager.toFullPropertyName(queue, QueueACL.ADMINISTER_JOBS.getAclName()), acl.getAclString());
TokenCache.cleanUpTokenReferral(conf);
if (conf.getBoolean("mapreduce.job.token.tracking.ids.enabled", false)) {
ArrayList<String> trackingIds = new ArrayList();
Iterator i$ = job.getCredentials().getAllTokens().iterator();
while(i$.hasNext()) {
Token<? extends TokenIdentifier> t = (Token)i$.next();
trackingIds.add(t.decodeIdentifier().getTrackingId());
}
conf.setStrings("mapreduce.job.token.tracking.ids", (String[])trackingIds.toArray(new String[trackingIds.size()]));
}
ReservationId reservationId = job.getReservationId();
if (reservationId != null) {
conf.set("mapreduce.job.reservation.id", reservationId.toString());
}
//將job的configuration信息寫入到 submitJobDir/job.xml
this.writeConf(conf, submitJobFile);
this.printTokens(jobId, job.getCredentials());
//通過client提交job,包括job資源目錄,驗證信息.
//這里要看使用的client是YARNRunner還是LocalRunner
//最后返回提交job的狀態(tài)
status = this.submitClient.submitJob(jobId, submitJobDir.toString(), job.getCredentials());
if (status == null) {
throw new IOException("Could not launch job");
}
var24 = status;
} finally {
//如果提交任務失敗,則刪除jobdir
if (status == null) {
LOG.info("Cleaning up the staging area " + submitJobDir);
if (this.jtFs != null && submitJobDir != null) {
this.jtFs.delete(submitJobDir, true);
}
}
}
return var24;
}
總結(jié)一下上面的主要流程:
(1)Path jobStagingArea = JobSubmissionFiles.getStagingDir(cluster, conf);
獲取job總的工作目錄
(2)JobID jobId = this.submitClient.getNewJobID();
job.setJobID(jobId);
通過處理client向集群申請jobid,并保持到job的配置信息中。
(3)Path submitJobDir = new Path(jobStagingArea, jobId.toString());
獲取當前job的工作目錄,以及jobid命名
(4)this.copyAndConfigureFiles(job, submitJobDir);
復制job的臨時文件,運行的jar包到submitJobDir下
(5)Path submitJobFile = JobSubmissionFiles.getJobConfPath(submitJobDir);
獲取job配置信息文件的路徑。命名為:submitJobDir/job.xml
(6)int maps = this.writeSplits(job, submitJobDir);
將切片信息存儲到submitJobDir下,并返回切片數(shù)目。會調(diào)用 InputFormat.getSplits()來獲取規(guī)劃的切片信息。切片信息會寫入到 submitJobDir/job.split,切片信息條目的元信息寫入到 submitJobDir/job.splitmetainfo。
(7)this.writeConf(conf, submitJobFile);
將job配置信息寫入到 submitJobDir/job.xml 中
(8)status = this.submitClient.submitJob(jobId, submitJobDir.toString(), job.getCredentials());
正式提交job,獲取job的提交狀態(tài)
下面挑比較復雜的看看這些的具體實現(xiàn)。
重點在于job任務的資源的生成,如切片文件的生成。
=================================================================
復制job的臨時文件,運行的jar包到submitJobDir下
//------------------JobSubmitter.java
private void copyAndConfigureFiles(Job job, Path jobSubmitDir) throws IOException {
JobResourceUploader rUploader = new JobResourceUploader(this.jtFs);
rUploader.uploadFiles(job, jobSubmitDir);
job.getWorkingDirectory();
}
//----------------------JobResourceUploader.java
public void uploadFiles(Job job, Path submitJobDir) throws IOException {
......................
String files = conf.get("tmpfiles");
String libjars = conf.get("tmpjars");
String archives = conf.get("tmparchives");
String jobJar = job.getJar();
..................代碼長,就截取一點,這些就是要復制到job目錄的文件類型
}
可以看到主要復制jar包以及相關(guān)的文件到job工作目錄下。
獲取job配置信息文件的路徑。命名為:submitJobDir/job.xml
//-----------------------------JobSubmissionFiles.java
public static Path getJobConfPath(Path jobSubmitDir) {
return new Path(jobSubmitDir, "job.xml");
}
將切片信息存儲到submitJobDir下,并返回切片數(shù)目。會調(diào)用 InputFormat.getSplits()來獲取規(guī)劃的切片信息。切片信息會寫入到 submitJobDir/job.split,切片信息條目的元信息寫入到 submitJobDir/job.splitmetainfo。返回的是切片數(shù)目
//------------------------JobSubmitter.java
private int writeSplits(JobContext job, Path jobSubmitDir) throws IOException, InterruptedException, ClassNotFoundException {
JobConf jConf = (JobConf)job.getConfiguration();
int maps;
if (jConf.getUseNewMapper()) {
maps = this.writeNewSplits(job, jobSubmitDir);
} else {
maps = this.writeOldSplits(jConf, jobSubmitDir);
}
return maps;
}
沒什么特別的,主要就是區(qū)分新舊api,我們看 this.writeNewSplits
//------------------------JobSubmitter.java
private <T extends InputSplit> int writeNewSplits(JobContext job, Path jobSubmitDir) throws IOException, InterruptedException, ClassNotFoundException {
Configuration conf = job.getConfiguration();
//反射獲取指定的inputformat對象,默認TextInputFormat
InputFormat<?, ?> input = (InputFormat)ReflectionUtils.newInstance(job.getInputFormatClass(), conf);
//通過inputformat的getSplits() 生成獲取規(guī)劃切片信息
List<InputSplit> splits = input.getSplits(job);
T[] array = (InputSplit[])((InputSplit[])splits.toArray(new InputSplit[splits.size()]));
Arrays.sort(array, new JobSubmitter.SplitComparator());
//創(chuàng)建切片文件原始數(shù)據(jù)文件,以及元數(shù)據(jù)文件
JobSplitWriter.createSplitFiles(jobSubmitDir, conf, jobSubmitDir.getFileSystem(conf), array);
return array.length;
}
獲取 inputformat對象,通過inputformat的getSplits() 獲取規(guī)劃切片信息,然后JobSplitWriter.createSplitFiles()創(chuàng)建切片信息文件。下面最后這個方法
//------------------JobSplitWriter.createSplitFiles
public static <T extends InputSplit> void createSplitFiles(Path jobSubmitDir, Configuration conf, FileSystem fs, T[] splits) throws IOException, InterruptedException {
//創(chuàng)建切片輸出流,文件命名為 jobSubmitDir/job.split
FSDataOutputStream out = createFile(fs, JobSubmissionFiles.getJobSplitFile(jobSubmitDir), conf);
//將數(shù)組中的每個切片元信息進行序列化,并將切片信息寫入到jobSubmitDir/job.split中
//返回的是每個切片條目的元信息,比如每條切片信息在 job.split中的起始位置,長度等
SplitMetaInfo[] info = writeNewSplits(conf, splits, out);
out.close();
//將切片信息文件的元信息寫入到文件 jobSubmitDir/job.splitmetainfo 中
writeJobSplitMetaInfo(fs, JobSubmissionFiles.getJobSplitMetaFile(jobSubmitDir), new FsPermission(JobSubmissionFiles.JOB_FILE_PERMISSION), 1, info);
}
這里主要生成兩個主要文件
jobSubmitDir/job.split:切片信息文件,記錄每個切片的信息,比如路徑,block位置,偏移量等
jobSubmitDir/job.splitmetainfo:切片信息文件中每個信息條目的索引位置,如每條切片信息在 job.split中的起始位置,長度等
下面看看這兩個文件的生成
首先是jobSubmitDir/job.split
private static <T extends InputSplit> SplitMetaInfo[] writeNewSplits(Configuration conf, T[] array, FSDataOutputStream out) throws IOException, InterruptedException {
SplitMetaInfo[] info = new SplitMetaInfo[array.length];
if (array.length != 0) {
SerializationFactory factory = new SerializationFactory(conf);
int i = 0;
int maxBlockLocations = conf.getInt("mapreduce.job.max.split.locations", 10);
long offset = out.getPos();
InputSplit[] arr$ = array;
int len$ = array.length;
//循環(huán)將切片信息中每一條切片信息寫入到文件中,并生成每條切片信息的元信息
for(int i$ = 0; i$ < len$; ++i$) {
T split = arr$[i$];
long prevCount = out.getPos();
Text.writeString(out, split.getClass().getName());
Serializer<T> serializer = factory.getSerializer(split.getClass());
serializer.open(out);
//將切片信息對象序列化存儲到文件中
serializer.serialize(split);
long currCount = out.getPos();
String[] locations = split.getLocations();
if (locations.length > maxBlockLocations) {
LOG.warn("Max block location exceeded for split: " + split + " splitsize: " + locations.length + " maxsize: " + maxBlockLocations);
locations = (String[])Arrays.copyOf(locations, maxBlockLocations);
}
//生成每條切片信息的元信息
info[i++] = new SplitMetaInfo(locations, offset, split.getLength());
offset += currCount - prevCount;
}
}
return info;
}
主要就是將split中的切片信息條目對象序列化寫入到文件中,并生成jobSubmitDir/job.splitmetainfo中要寫入的信息,也就是切片文件的索引信息
接著看看 writeJobSplitMetaInfo()
private static void writeJobSplitMetaInfo(FileSystem fs, Path filename, FsPermission p, int splitMetaInfoVersion, SplitMetaInfo[] allSplitMetaInfo) throws IOException {
//寫入切片信息條目的元信息,創(chuàng)建一個輸出流
FSDataOutputStream out = FileSystem.create(fs, filename, p);
out.write(JobSplit.META_SPLIT_FILE_HEADER);
WritableUtils.writeVInt(out, splitMetaInfoVersion);
WritableUtils.writeVInt(out, allSplitMetaInfo.length);
SplitMetaInfo[] arr$ = allSplitMetaInfo;
int len$ = allSplitMetaInfo.length;
//逐條寫入
for(int i$ = 0; i$ < len$; ++i$) {
SplitMetaInfo splitMetaInfo = arr$[i$];
splitMetaInfo.write(out);
}
out.close();
}
這里其實很明顯了,就是將切片文件索引信息寫入到 jobSubmitDir/job.splitmetainfo
正式提交job,獲取job的提交狀態(tài)
public JobStatus submitJob(JobID jobId, String jobSubmitDir, Credentials ts) throws IOException, InterruptedException {
this.addHistoryToken(ts);
//這里就是將job配置,以及job資源的hdfs目錄路徑傳入
ApplicationSubmissionContext appContext = this.createApplicationSubmissionContext(this.conf, jobSubmitDir, ts);
try {
//提交job,返回的appid
ApplicationId applicationId = this.resMgrDelegate.submitApplication(appContext);
//根據(jù)appid創(chuàng)建appMaster
ApplicationReport appMaster = this.resMgrDelegate.getApplicationReport(applicationId);
String diagnostics = appMaster == null ? "application report is null" : appMaster.getDiagnostics();
if (appMaster != null && appMaster.getYarnApplicationState() != YarnApplicationState.FAILED && appMaster.getYarnApplicationState() != YarnApplicationState.KILLED) {
return this.clientCache.getClient(jobId).getJobStatus(jobId);
} else {
throw new IOException("Failed to run job : " + diagnostics);
}
} catch (YarnException var8) {
throw new IOException(var8);
}
}
這里主要就是提交job,創(chuàng)建appMaster。最后獲取job狀態(tài)。
一個job提交流程主要如下:
1、和MapReduce集群建立連接 this.connect()
這里面最重要就是創(chuàng)建了 client,有 YARNRunner和LocalJobRunner兩種方式。后續(xù)用來和server端通信、提交job等。
2、正式提交job ,submitter.submitJobInternal(Job.this, cluster)
(1)Path jobStagingArea = JobSubmissionFiles.getStagingDir(cluster, conf);
獲取job總的工作目錄
(2)JobID jobId = this.submitClient.getNewJobID();
job.setJobID(jobId);
通過處理client向集群申請jobid,并保持到job的配置信息中。
(3)Path submitJobDir = new Path(jobStagingArea, jobId.toString());
獲取當前job的工作目錄,以及jobid命名
(4)this.copyAndConfigureFiles(job, submitJobDir);
復制job的臨時文件,運行的jar包到submitJobDir下
(5)Path submitJobFile = JobSubmissionFiles.getJobConfPath(submitJobDir);
獲取job配置信息文件的路徑。命名為:submitJobDir/job.xml
(6)int maps = this.writeSplits(job, submitJobDir);
將切片信息存儲到submitJobDir下,并返回切片數(shù)目。會調(diào)用 InputFormat.getSplits()來獲取規(guī)劃的切片信息。切片信息會寫入到 submitJobDir/job.split,切片信息條目的元信息寫入到 submitJobDir/job.splitmetainfo。
(7)this.writeConf(conf, submitJobFile);
將job配置信息寫入到 submitJobDir/job.xml 中
(8)status = this.submitClient.submitJob(jobId, submitJobDir.toString(), job.getCredentials());
正式提交job,獲取job的提交狀態(tài)
免責聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進行舉報,并提供相關(guān)證據(jù),一經(jīng)查實,將立刻刪除涉嫌侵權(quán)內(nèi)容。