1 常用輸入格式
| 輸入格式 | 特點(diǎn) | 使用的RecordReader | 是否使用FileInputFormat的getSplits |
|---|---|---|---|
| TextInputFormat | 以行偏移量為key，以換行符前的字符為Value | LineRecordReader | 是 |
| KeyValueTextInputFormat | 默認(rèn)分割符為”\t”肾扰，根據(jù)分割符來切分行壤短，前為key噩峦，后為value | KeyValueLineRecordReader，內(nèi)部使用LineRecordReader | 是 |
| NLineInputFormat | 根據(jù)屬性mapreduce.input.lineinputformat.linespermap所設(shè)置的行數(shù)為每片split的行數(shù) | LineRecordReader | 覆蓋FileInputFormat的getSplits |
| SequenceFileInputFormat | 使用Hadoop特有文件格式SequenceFile.Reader進(jìn)行讀寫残炮，讀取二進(jìn)制文件 | SequenceFileRecordReader | 是 |
| DBInputFormat | 通過與數(shù)據(jù)建立連接屋休，將讀取的數(shù)據(jù)根據(jù)map數(shù)進(jìn)行分片 | DBRecordReader | 繼承InputFormat，實(shí)現(xiàn)分片和RecordReader |
InputFormat層級(jí).JPG
2 自定義InputFormat的流程
1）如果是文本格式的數(shù)據(jù)梆惯，那么實(shí)現(xiàn)一個(gè)XXInputForamt繼承FileInputFormat
2）重寫 FileInputFormat 里面的 isSplitable() 方法。如果文件是壓縮文件的話則不能切割吗垮，一般都是支持切割
3）重寫 FileInputFormat 里面的 createRecordReader()方法
4）自定義XXRecordReader垛吗，來讀取特定的格式
XXRecordReader中需要重點(diǎn)實(shí)現(xiàn)以下兩個(gè)的方法
@Override
public void initialize(InputSplit input, TaskAttemptContext context)
throws IOException, InterruptedException {
FileSplit split=(FileSplit)input;
Configuration job=context.getConfiguration();
Path file=split.getPath();
FileSystem fs=file.getFileSystem(job);
FSDataInputStream fileIn=fs.open(file);
//紅色標(biāo)記這部分對(duì)于文本型數(shù)據(jù)來說基本是一樣的
in=new LineReader(fileIn,job);
line=new Text();
lineKey=new Text();
lineValue = new Text();
}
//此方法讀取每行數(shù)據(jù)，完成自定義的key和value
@Override
public boolean nextKeyValue() throws IOException, InterruptedException {
int linesize=in.readLine(line);//每行數(shù)據(jù)
if(linesize==0) return false;
String[] pieces = line.toString().split("\\s+");//解析每行數(shù)據(jù)
...
lineKey.set(“key”);//完成自定義key數(shù)據(jù)
lineValue.set(“value”);//封裝自定義value數(shù)據(jù)
return true;
}
3 多個(gè)輸入
1）如果輸入格式存在多種烁登，可以設(shè)置不同Mapper處理不同的數(shù)據(jù)源
MultipleInputs.addInputPath(job,ncdcInputPath,TextInputFormat.class,NCDCTemperatureMapper.class);
2）存在多種輸入格式怯屉，而只有一個(gè)Mapper則可使用
public static void addInputPath(Job job,Path path,class< ? extends InputFormat> inputFormatClass);
使用job.setMapperClass();
4. InputFormat及其子類解決的是針對(duì)不同的數(shù)據(jù)格式分片和讀取問題
4.1 getSplits方法中實(shí)現(xiàn)如何分片?
1）根據(jù)不同的數(shù)據(jù)來源采取不同的切片方式
例子1：文本格式的數(shù)據(jù)來源
通過計(jì)算確認(rèn)splitSize的大小蔚舀，假如輸入文件為100M，那么splitSize則為64M锨络，那么文件會(huì)被切分為64M和36M兩個(gè)分片輸出赌躺。
public List<InputSplit> getSplits(JobContext job) throws IOException {
StopWatch sw = new StopWatch().start();
long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
long maxSize = getMaxSplitSize(job);
// generate splits
List<InputSplit> splits = new ArrayList<InputSplit>();
List<FileStatus> files = listStatus(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)) {
long blockSize = file.getBlockSize();
long splitSize = computeSplitSize(blockSize, minSize, maxSize);
long bytesRemaining = length;
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
if (LOG.isDebugEnabled()) {
// Log only if the file is big enough to be splitted
if (length > Math.min(file.getBlockSize(), minSize)) {
LOG.debug("File is not splittable so no parallelization "
+ "is possible: " + file.getPath());
}
}
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;
}
例子2：數(shù)據(jù)來源為數(shù)據(jù)庫
通過查詢需要讀取的條數(shù)，然后再除以map_task數(shù)羡儿，得出每個(gè)map_task需要處理的條數(shù)礼患，進(jìn)行切分
public List<InputSplit> getSplits(JobContext job) throws IOException {
ResultSet results = null;
Statement statement = null;
try {
statement = connection.createStatement();
results = statement.executeQuery(getCountQuery());//查詢總條數(shù)
results.next();
long count = results.getLong(1);
int chunks = job.getConfiguration().getInt(MRJobConfig.NUM_MAPS, 1);
long chunkSize = (count / chunks);
results.close();
statement.close();
List<InputSplit> splits = new ArrayList<InputSplit>();
// Split the rows into n-number of chunks and adjust the last chunk
// accordingly
for (int i = 0; i < chunks; i++) {
DBInputSplit split;
if ((i + 1) == chunks)
split = new DBInputSplit(i * chunkSize, count);
else
split = new DBInputSplit(i * chunkSize, (i * chunkSize)
+ chunkSize);
splits.add(split);
}
connection.commit();
return splits;
} catch (SQLException e) {
throw new IOException("Got SQLException", e);
} finally {
try {
if (results != null) { results.close(); }
} catch (SQLException e1) {}
try {
if (statement != null) { statement.close(); }
} catch (SQLException e1) {}
closeConnection();
}
}
2）RecordReader類實(shí)現(xiàn)如何讀取數(shù)據(jù)
一般的文本格式一般使用LineRecordReader進(jìn)行讀取，然后根據(jù)需求進(jìn)行處理
