2. MapReduce中的数据输入
2.1 文件切片
2.1.1 什么是切片
数据块(Block):HDFS中数据保存的单位,HDFS在物理上将数据分为一个一个Block管理
数据切片(Split):在逻辑上对Map任务输入数据的切片。
2.1.2 为什么要切片
将输入文件分为多片可以并行进行Map阶段的计算,提高Job的运行速度。一份数据切片就会有一个MapTask。
2.1.3 文件的切片机制
- 简单的按照文件的内容长度切片,切片大小默认为Block的大小(128M),但每次切片完都会判断剩下的部分是否大于块的1.1倍,不大于1.1倍就归入上一个切片
- 切片时不会考虑数据集整体,而是单独针对每一份文件切片(默认)
2.2 任务提交流程
//在客户端Driver中提交任务
job.waitForCompletion()
submit();
// 1建立连接
connect();
// 1)创建提交Job的代理
new Cluster(getConfiguration());
// (1)判断是本地yarn还是远程
initialize(jobTrackAddr, conf);
// 2 提交job
submitter.submitJobInternal(Job.this, cluster)
// 1)创建给集群提交数据的Stag路径
Path jobStagingArea = JobSubmissionFiles.getStagingDir(cluster, conf);
// 2)获取jobid ,并创建Job路径
JobID jobId = submitClient.getNewJobID();
// 3)拷贝jar包到集群
copyAndConfigureFiles(job, submitJobDir);
rUploader.uploadFiles(job, jobSubmitDir);
// 4)计算切片,生成切片规划文件
writeSplits(job, submitJobDir);
maps = writeNewSplits(job, jobSubmitDir);
input.getSplits(job);
long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
long maxSize = getMaxSplitSize(job);
for (FileStatus file: files)
long splitSize = computeSplitSize(blockSize, minSize, maxSize);
//blockSize与给定的最大值的最小值与给定的Split最小值取最大值
//最小值默认为1,最大值默认为Long.MAX_VALUE
Math.max(minSize, Math.min(maxSize, blockSize));
// 5)向Stag路径写XML配置文件
writeConf(conf, submitJobFile);
conf.writeXml(out);
// 6)提交Job,返回提交状态
status = submitClient.submitJob(jobId, submitJobDir.toString(), job.getCredentials());
![1586359954724](picture\submit.png)
- Job.submit运行,获得一个JobSubmitter对象;
- 找到数据存储的目录,获取JobId等信息;
- 遍历所有文件,按要求规划切片信息,并将切片信息写入job.split文件中(只包含了元数据信息)
- 获取相关参数的.xml文件和任务jar包;
- 提交到YARN,根据切片信息分配MapTask。
如何决定Map和Reduce的数量
1) map的数量:splitSize=max(minSize,min{maxSize,blockSize}) map个数由切片数决定
2) reduce的数量:job.setNumReduceTasks(x);默认为1
2.3 几种不同的切片方法
TextInputFormat(默认)
不管文件多小,都会单独规划切片,如果有大量小文件,就会产生大量的MapTask,处理效率极其低下。
CombineTextInputFormat
会将多个小文件从逻辑上规划到一个切片中。
CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);// 4m
2.4 MapTask读取数据的方法(FileInputFormat)
MapTask将文件读取如内存并按K-V对的形式保存,具体实现依靠FileInputFormat的实现类
2.4.1 TextInputFormat(默认)
按行读取:
键是该行起始位置在整个文件中字节偏移量,LongWritable类型;
值是这行内容,不包括换行符和回车符,Text类型。
2.4.2 KeyValueTextInputFormat
按行读取:
被分隔符分割为K-V对,默认分隔符为’\t’。
2.4.3 NlineInputFormat
文件分片方式不同:每个MapTask不再按Block划分,而是按NlineInputFormat指定的行数N来划分。
键值对读取与默认相同
2.4.4 自定义InputFormat实现类
案例:可以通过这种方法读取小文件合并为一个SequenceFile文件实现大量小文件的合并。
流程:
- 自定义类继承FileInputFormat;
- 重写isSplitable()返回false不可分割;
- 重写CreateRecordReader,创建自定义的Reader对象;
- 改写RecordReader,实现一次读取一个文件,封装为KV对;
- IO流一次读入一个文件,保存为value;
- 文件路径信息+文件名,保存为key;
// 定义类继承FileInputFormat
public class WholeFileInputformat extends FileInputFormat<Text, BytesWritable>{
@Override
protected boolean isSplitable(JobContext context, Path filename) {
return false;
}
@Override
public RecordReader<Text, BytesWritable> createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
WholeRecordReader recordReader = new WholeRecordReader();
recordReader.initialize(split, context);
return recordReader;
}
}
public class WholeRecordReader extends RecordReader<Text, BytesWritable>{
private Configuration configuration;
private FileSplit split;
private boolean isProgress= true;
private BytesWritable value = new BytesWritable();
private Text k = new Text();
@Override
public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
this.split = (FileSplit)split;
configuration = context.getConfiguration();
}
@Override
public boolean nextKeyValue() throws IOException, InterruptedException {
if (isProgress) {
// 1 定义缓存区
byte[] contents = new byte[(int)split.getLength()];
FileSystem fs = null;
FSDataInputStream fis = null;
try {
// 2 获取文件系统
Path path = split.getPath();
fs = path.getFileSystem(configuration);
// 3 读取数据
fis = fs.open(path);
// 4 读取文件内容
IOUtils.readFully(fis, contents, 0, contents.length);
// 5 输出文件内容
value.set(contents, 0, contents.length);
// 6 获取文件路径及名称
String name = split.getPath().toString();
// 7 设置输出的key值
k.set(name);
} catch (Exception e) {
}finally {
IOUtils.closeStream(fis);
}
isProgress = false;
return true;
}
return false;
}
@Override
public Text getCurrentKey() throws IOException, InterruptedException {
return k;
}
@Override
public BytesWritable getCurrentValue() throws IOException, InterruptedException {
return value;
}
@Override
public float getProgress() throws IOException, InterruptedException {
return 0;
}
@Override
public void close() throws IOException {
}
}
public class SequenceFileMapper extends Mapper<Text, BytesWritable, Text, BytesWritable>{
@Override
protected void map(Text key, BytesWritable value,Context context)throws IOException, InterruptedException {
context.write(key, value);
}
}
public class SequenceFileReducer extends Reducer<Text, BytesWritable, Text, BytesWritable> {
@Override
protected void reduce(Text key, Iterable<BytesWritable> values, Context context) throws IOException, InterruptedException {
context.write(key, values.iterator().next());
}
}
public class SequenceFileDriver {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
// 输入输出路径需要根据自己电脑上实际的输入输出路径设置
args = new String[] { "e:/input/inputinputformat", "e:/output1" };
// 1 获取job对象
Configuration conf = new Configuration();
Job job = Job.getInstance(conf);
// 2 设置jar包存储位置、关联自定义的mapper和reducer
job.setJarByClass(SequenceFileDriver.class);
job.setMapperClass(SequenceFileMapper.class);
job.setReducerClass(SequenceFileReducer.class);
// 7设置输入的inputFormat
job.setInputFormatClass(WholeFileInputformat.class);
// 8设置输出的outputFormat
job.setOutputFormatClass(SequenceFileOutputFormat.class);
// 3 设置map输出端的kv类型
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(BytesWritable.class);
// 4 设置最终输出端的kv类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(BytesWritable.class);
// 5 设置输入输出路径
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
// 6 提交job
boolean result = job.waitForCompletion(true);
System.exit(result ? 0 : 1);
}
}