MapReduce之Job提交流程源碼和切片源碼分析
hadoop2.7.2 MapReduce Job提交源碼及切片源碼分析node
1. 首先從
waitForCompletion函數進入web
boolean result = job.waitForCompletion(true);複製代碼
/**
* Submit the job to the cluster and wait for it to finish.
* @param verbose print the progress to the user
* @return true if the job succeeded
* @throws IOException thrown if the communication with the
* <code>JobTracker</code> is lost
*/
public boolean waitForCompletion(boolean verbose
) throws IOException, InterruptedException,
ClassNotFoundException {
// 首先判斷state，當state爲DEFINE時能夠提交，進入 submit() 方法
if (state == JobState.DEFINE) {
submit();
}
if (verbose) {
monitorAndPrintJob();
} else {
// get the completion poll interval from the client.
int completionPollIntervalMillis =
Job.getCompletionPollInterval(cluster.getConf());
while (!isComplete()) {
try {
Thread.sleep(completionPollIntervalMillis);
} catch (InterruptedException ie) {
}
}
}
return isSuccessful();
}複製代碼
2. 進入
submit()方法apache
/**
* Submit the job to the cluster and return immediately.
* @throws IOException
*/
public void submit()
throws IOException, InterruptedException, ClassNotFoundException {
// 確認JobState狀態爲可提交狀態，不然不能提交
ensureState(JobState.DEFINE);
// 設置使用最新的API
setUseNewAPI();
// 進入connect()方法，MapReduce做業提交時鏈接集羣是經過Job類的connect()方法實現的，
// 它其實是構造集羣Cluster實例cluster
connect();
// connect()方法執行完以後，定義提交者submitter
final JobSubmitter submitter =
getJobSubmitter(cluster.getFileSystem(), cluster.getClient());
status = ugi.doAs(new PrivilegedExceptionAction<JobStatus>() {
public JobStatus run() throws IOException, InterruptedException,
ClassNotFoundException {
// 這裏的核心方法是submitJobInternal(),顧名思義，提交job的內部方法，實現了提交job的全部業務邏輯
// 進入submitJobInternal
return submitter.submitJobInternal(Job.this, cluster);
}
});
// 提交以後state狀態改變
state = JobState.RUNNING;
LOG.info("The url to track the job: " + getTrackingURL());
}複製代碼
3. 進入
connect()方法緩存
- MapReduce做業提交時鏈接集羣經過Job的Connect方法實現，它其實是構造集羣Cluster實例cluster
- cluster是鏈接MapReduce集羣的一種工具，提供了獲取MapReduce集羣信息的方法
- 在Cluster內部，有一個與集羣進行通訊的客戶端通訊協議ClientProtocol的實例client，它由ClientProtocolProvider的靜態create()方法構造
- 在create內部，Hadoop2.x中提供了兩種模式的ClientProtocol，分別爲Yarn模式的YARNRunner和Local模式的LocalJobRunner，Cluster其實是由它們負責與集羣進行通訊的
private synchronized void connect()
throws IOException, InterruptedException, ClassNotFoundException {
if (cluster == null) {// cluster提供了遠程獲取MapReduce的方法
cluster =
ugi.doAs(new PrivilegedExceptionAction<Cluster>() {
public Cluster run()
throws IOException, InterruptedException,
ClassNotFoundException {
// 只需關注這個Cluster()構造器，構造集羣cluster實例
return new Cluster(getConfiguration());
}
});
}
}複製代碼
4. 進入
Cluster()構造器框架
// 首先調用一個參數的構造器，間接調用兩個參數的構造器
public Cluster(Configuration conf) throws IOException {
this(null, conf);
}
public Cluster(InetSocketAddress jobTrackAddr, Configuration conf)
throws IOException {
this.conf = conf;
this.ugi = UserGroupInformation.getCurrentUser();
// 最重要的initialize方法
initialize(jobTrackAddr, conf);
}
// cluster中要關注的兩個成員變量是客戶端通信協議提供者ClientProtocolProvider和客戶端通信協議ClientProtocol實例client
private void initialize(InetSocketAddress jobTrackAddr, Configuration conf)
throws IOException {
synchronized (frameworkLoader) {
for (ClientProtocolProvider provider : frameworkLoader) {
LOG.debug("Trying ClientProtocolProvider : "
+ provider.getClass().getName());
ClientProtocol clientProtocol = null;
try {
// 若是配置文件沒有配置YARN信息，則構建LocalRunner,MR任務本地運行
// 若是配置文件有配置YARN信息，則構建YarnRunner,MR任務在YARN集羣上運行
if (jobTrackAddr == null) {
// 客戶端通信協議client是調用ClientProtocolProvider的create()方法實現
clientProtocol = provider.create(conf);
} else {
clientProtocol = provider.create(jobTrackAddr, conf);
}
if (clientProtocol != null) {
clientProtocolProvider = provider;
client = clientProtocol;
LOG.debug("Picked " + provider.getClass().getName()
+ " as the ClientProtocolProvider");
break;
}
else {
LOG.debug("Cannot pick " + provider.getClass().getName()
+ " as the ClientProtocolProvider - returned null protocol");
}
}
catch (Exception e) {
LOG.info("Failed to use " + provider.getClass().getName()
+ " due to error: ", e);
}
}
}
if (null == clientProtocolProvider || null == client) {
throw new IOException(
"Cannot initialize Cluster. Please check your configuration for "
+ MRConfig.FRAMEWORK_NAME
+ " and the correspond server addresses.");
}
}
複製代碼
5. 進入
submitJobInternal()，job的內部提交方法，用於提交job到集羣分佈式
JobStatus submitJobInternal(Job job, Cluster cluster)
throws ClassNotFoundException, InterruptedException, IOException {
//validate the jobs output specs
// 檢查結果的輸出路徑是否已經存在，若是存在會報異常
checkSpecs(job);
// conf裏邊是集羣的xml配置文件信息
Configuration conf = job.getConfiguration();
// 添加MR框架到分佈式緩存中
addMRFrameworkToDistributedCache(conf);
// 獲取提交執行時相關資源的臨時存放路徑
// 參數未配置時默認是（工做空間根目錄下的）/tmp/hadoop-yarn/staging/提交做業用戶名/.staging
Path jobStagingArea = JobSubmissionFiles.getStagingDir(cluster, conf);
//configure the command line options correctly on the submitting dfs
InetAddress ip = InetAddress.getLocalHost();
if (ip != null) {//記錄提交做業的主機IP、主機名，而且設置配置信息conf
submitHostAddress = ip.getHostAddress();
submitHostName = ip.getHostName();
conf.set(MRJobConfig.JOB_SUBMITHOST,submitHostName);
conf.set(MRJobConfig.JOB_SUBMITHOSTADDR,submitHostAddress);
}
// 獲取JobId
JobID jobId = submitClient.getNewJobID();
// 設置jobId
job.setJobID(jobId);
// 提交做業的路徑Path(Path parent, String child)，會將兩個參數拼接爲一個路徑
Path submitJobDir = new Path(jobStagingArea, jobId.toString());
// job的狀態
JobStatus status = null;
try {
conf.set(MRJobConfig.USER_NAME,
UserGroupInformation.getCurrentUser().getShortUserName());
conf.set("hadoop.http.filter.initializers",
"org.apache.hadoop.yarn.server.webproxy.amfilter.AmFilterInitializer");
conf.set(MRJobConfig.MAPREDUCE_JOB_DIR, submitJobDir.toString());
LOG.debug("Configuring job " + jobId + " with " + submitJobDir
+ " as the submit dir");
// get delegation token for the dir
TokenCache.obtainTokensForNamenodes(job.getCredentials(),
new Path[] { submitJobDir }, conf);
populateTokenCache(conf, job.getCredentials());
// generate a secret to authenticate shuffle transfers
if (TokenCache.getShuffleSecretKey(job.getCredentials()) == null) {
KeyGenerator keyGen;
try {
keyGen = KeyGenerator.getInstance(SHUFFLE_KEYGEN_ALGORITHM);
keyGen.init(SHUFFLE_KEY_LENGTH);
} catch (NoSuchAlgorithmException e) {
throw new IOException("Error generating shuffle secret key", e);
}
SecretKey shuffleKey = keyGen.generateKey();
TokenCache.setShuffleSecretKey(shuffleKey.getEncoded(),
job.getCredentials());
}
if (CryptoUtils.isEncryptedSpillEnabled(conf)) {
conf.setInt(MRJobConfig.MR_AM_MAX_ATTEMPTS, 1);
LOG.warn("Max job attempts set to 1 since encrypted intermediate" +
"data spill is enabled");
}
// 拷貝jar包到集羣
// 此方法中調用以下方法：rUploader.uploadFiles(job, jobSubmitDir);
// uploadFiles方法將jar包拷貝到集羣
copyAndConfigureFiles(job, submitJobDir);
Path submitJobFile = JobSubmissionFiles.getJobConfPath(submitJobDir);
// Create the splits for the job
LOG.debug("Creating splits at " + jtFs.makeQualified(submitJobDir));
// 計算切片，生成切片規劃文件
int maps = writeSplits(job, submitJobDir);
conf.setInt(MRJobConfig.NUM_MAPS, maps);
LOG.info("number of splits:" + maps);
// write "queue admins of the queue to which job is being submitted"
// to job file.
String queue = conf.get(MRJobConfig.QUEUE_NAME,
JobConf.DEFAULT_QUEUE_NAME);
AccessControlList acl = submitClient.getQueueAdmins(queue);
conf.set(toFullPropertyName(queue,
QueueACL.ADMINISTER_JOBS.getAclName()), acl.getAclString());
// removing jobtoken referrals before copying the jobconf to HDFS
// as the tasks don't need this setting, actually they may break
// because of it if present as the referral will point to a
// different job.
TokenCache.cleanUpTokenReferral(conf);
if (conf.getBoolean(
MRJobConfig.JOB_TOKEN_TRACKING_IDS_ENABLED,
MRJobConfig.DEFAULT_JOB_TOKEN_TRACKING_IDS_ENABLED)) {
// Add HDFS tracking ids
ArrayList<String> trackingIds = new ArrayList<String>();
for (Token<? extends TokenIdentifier> t :
job.getCredentials().getAllTokens()) {
trackingIds.add(t.decodeIdentifier().getTrackingId());
}
conf.setStrings(MRJobConfig.JOB_TOKEN_TRACKING_IDS,
trackingIds.toArray(new String[trackingIds.size()]));
}
// Set reservation info if it exists
ReservationId reservationId = job.getReservationId();
if (reservationId != null) {
conf.set(MRJobConfig.RESERVATION_ID, reservationId.toString());
}
// Write job file to submit dir
writeConf(conf, submitJobFile);
//
// Now, actually submit the job (using the submit name)
// 開始正式提交job
printTokens(jobId, job.getCredentials());
status = submitClient.submitJob(
jobId, submitJobDir.toString(), job.getCredentials());
if (status != null) {
return status;
} else {
throw new IOException("Could not launch job");
}
} finally {
if (status == null) {
LOG.info("Cleaning up the staging area " + submitJobDir);
if (jtFs != null && submitJobDir != null)
jtFs.delete(submitJobDir, true);
}
}
}複製代碼
6. 進入
writeSplits(job, submitJobDir)，計算切片，生成切片規劃文件ide
- 內部會調用
writeNewSplits(job, jobSubmitDir)方法 writeNewSplits(job, jobSubmitDir)內部定義了一個InputFormat類型的實例input- InputFormat主要做用：
- 驗證job的輸入規範
- 對輸入的文件進行切分，造成多個InputSplit（切片）文件，每個InputSplit對應着一個map任務（MapTask）
- 將切片後的數據按照規則造成key，value鍵值對RecordReader
- input調用getSplits（）方法：
Listsplits = input.getSplits(job);
7. 進入FileInputFormat類下的
getSplits(job)方法函數
/**
* Generate the list of files and make them into FileSplits.
* @param job the job context
* @throws IOException
*/
public List<InputSplit> getSplits(JobContext job) throws IOException {
StopWatch sw = new StopWatch().start();
// getFormatMinSplitSize()返回值固定爲1，getMinSplitSize(job)返回job大小
long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
// getMaxSplitSize(job)返回Lang類型的最大值
long maxSize = getMaxSplitSize(job);
// generate splits 生成切片
List<InputSplit> splits = new ArrayList<InputSplit>();
List<FileStatus> files = listStatus(job);
// 遍歷job下的全部文件
for (FileStatus file: files) {
// 獲取文件路徑
Path path = file.getPath();
// 獲取文件大小
long length = file.getLen();
if (length != 0) {
BlockLocation[] blkLocations;
if (file instanceof LocatedFileStatus) {
blkLocations = ((LocatedFileStatus) file).getBlockLocations();
} else {
FileSystem fs = path.getFileSystem(job.getConfiguration());
blkLocations = fs.getFileBlockLocations(file, 0, length);
}
// 判斷是否可分割
if (isSplitable(job, path)) {
// 獲取塊大小
// 本地環境塊大小默認爲32MB，YARN環境在hadoop2.x新版本爲128MB，舊版本爲64MB
long blockSize = file.getBlockSize();
// 計算切片的邏輯大小，默認等於塊大小
// 返回值爲：return Math.max(minSize, Math.min(maxSize, blockSize));
// 其中minSize=1， maxSize=Long類型最大值， blockSize爲切片大小
long splitSize = computeSplitSize(blockSize, minSize, maxSize);
long bytesRemaining = length;
// 每次切片時就要判斷切片剩下的部分是否大於切片大小的SPLIT_SLOP（默認爲1.1）倍，
// 不然就再也不切分，劃爲一塊
while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
splits.add(makeSplit(path, length-bytesRemaining, splitSize,
blkLocations[blkIndex].getHosts(),
blkLocations[blkIndex].getCachedHosts()));
bytesRemaining -= splitSize;
}
if (bytesRemaining != 0) {
int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
splits.add(makeSplit(path, length-bytesRemaining, bytesRemaining,
blkLocations[blkIndex].getHosts(),
blkLocations[blkIndex].getCachedHosts()));
}
} else { // not splitable
splits.add(makeSplit(path, 0, length, blkLocations[0].getHosts(),
blkLocations[0].getCachedHosts()));
}
} else {
//Create empty hosts array for zero length files
splits.add(makeSplit(path, 0, length, new String[0]));
}
}
// Save the number of input files for metrics/loadgen
job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());
sw.stop();
if (LOG.isDebugEnabled()) {
LOG.debug("Total # of splits generated by getSplits: " + splits.size()
+ ", TimeTaken: " + sw.now(TimeUnit.MILLISECONDS));
}
return splits;
}複製代碼
相關文章
- 1. MapReducejob提交流程源碼和切片源碼解析
- 2. MapReduce之job提交過程源碼分析-03
- 3. 【零】Hadoop中Mapreduce的Job默認切片流程源碼解析
- 4. 【零】Hadoop中Mapreduce的Job任務提交流程源碼解析
- 5. MapReduce之Job提交流程
- 6. spark job提交流程源碼
- 7. mapreduce 提交job 流程分析
- 8. MapReduce任務提交源碼分析
- 9. MapReduce做業提交源碼分析
- 10. MapReduce源碼分析——MapTask流程分析
- 更多相關文章...
- • Docker 資源彙總 - Docker教程
- • Java操作Neo4j數據庫（附帶源碼） - NoSQL教程
- • 互聯網組織的未來：剖析GitHub員工的任性之源
- • Java Agent入門實戰（二）-Instrumentation源碼概述
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
歡迎關注本站公眾號,獲取更多信息
