Tom White - Hadoop The Definitive Guide_ 4 edition - 2015 (811394), страница 47
Текст из файла (страница 47)
With the standard TextOutputFormat, thisrecord would be written to the output file with a tab separating a and b. You can changethe separator that TextOutputFormat uses by setting mapreduce.output.textoutputformat.separator.Table 8-3. Streaming separator propertiesProperty nameTypeDefault value Descriptionstream.map.input.field.separatorString \tThe separator to use when passing the input key and valuestrings to the stream map process as a stream of bytesstream.map.output.field.separatorString \tThe separator to use when splitting the output from the streammap process into key and value strings for the map outputstream.num.map.output.key.fieldsintThe number of fields separated by1stream.map.output.field.separatorto treat as the map output keystream.reduce.input.field.separatorString \tThe separator to use when passing the input key and valuestrings to the stream reduce process as a stream of bytesstream.reduce.output.field.separatorString \tThe separator to use when splitting the output from the streamreduce process into key and value strings for the final reduceoutputstream.num.reduce.output.key.fieldsintThe number of fields separated by1stream.reduce.output.field.separatorto treat as the reduce output keyMapReduce Types|219Figure 8-1.
Where separators are used in a Streaming MapReduce jobInput FormatsHadoop can process many different types of data formats, from flat text files to databases.In this section, we explore the different formats available.Input Splits and RecordsAs we saw in Chapter 2, an input split is a chunk of the input that is processed by a singlemap. Each map processes a single split. Each split is divided into records, and the mapprocesses each record—a key-value pair—in turn. Splits and records are logical: thereis nothing that requires them to be tied to files, for example, although in their mostcommon incarnations, they are.
In a database context, a split might correspond to arange of rows from a table and a record to a row in that range (this is precisely the casewith DBInputFormat, which is an input format for reading data from a relationaldatabase).Input splits are represented by the Java class InputSplit (which, like all of the classesmentioned in this section, is in the org.apache.hadoop.mapreduce package):1public abstract class InputSplit {public abstract long getLength() throws IOException, InterruptedException;public abstract String[] getLocations() throws IOException,InterruptedException;}An InputSplit has a length in bytes and a set of storage locations, which are just host‐name strings.
Notice that a split doesn’t contain the input data; it is just a reference tothe data. The storage locations are used by the MapReduce system to place map tasksas close to the split’s data as possible, and the size is used to order the splits so that the1. But see the classes in org.apache.hadoop.mapred for the old MapReduce API counterparts.220|Chapter 8: MapReduce Types and Formatslargest get processed first, in an attempt to minimize the job runtime (this is an instanceof a greedy approximation algorithm).As a MapReduce application writer, you don’t need to deal with InputSplits directly,as they are created by an InputFormat (an InputFormat is responsible for creating theinput splits and dividing them into records).
Before we see some concrete examples ofInputFormats, let’s briefly examine how it is used in MapReduce. Here’s the interface:public abstract class InputFormat<K, V> {public abstract List<InputSplit> getSplits(JobContext context)throws IOException, InterruptedException;public abstract RecordReader<K, V>createRecordReader(InputSplit split, TaskAttemptContext context)throws IOException, InterruptedException;}The client running the job calculates the splits for the job by calling getSplits(), thensends them to the application master, which uses their storage locations to schedulemap tasks that will process them on the cluster.
The map task passes the split to thecreateRecordReader() method on InputFormat to obtain a RecordReader for thatsplit. A RecordReader is little more than an iterator over records, and the map task usesone to generate record key-value pairs, which it passes to the map function. We can seethis by looking at the Mapper’s run() method:public void run(Context context) throws IOException, InterruptedException {setup(context);while (context.nextKeyValue()) {map(context.getCurrentKey(), context.getCurrentValue(), context);}cleanup(context);}After running setup(), the nextKeyValue() is called repeatedly on the Context (whichdelegates to the identically named method on the RecordReader) to populate the keyand value objects for the mapper.
The key and value are retrieved from the RecordReader by way of the Context and are passed to the map() method for it to do its work. Whenthe reader gets to the end of the stream, the nextKeyValue() method returns false,and the map task runs its cleanup() method and then completes.Input Formats|221Although it’s not shown in the code snippet, for reasons of efficien‐cy, RecordReader implementations will return the same key andvalue objects on each call to getCurrentKey() and getCurrentValue().
Only the contents of these objects are changed by the read‐er’s nextKeyValue() method. This can be a surprise to users, whomight expect keys and values to be immutable and not to be reused.This causes problems when a reference to a key or value object isretained outside the map() method, as its value can change withoutwarning. If you need to do this, make a copy of the object you wantto hold on to.
For example, for a Text object, you can use its copyconstructor: new Text(value).The situation is similar with reducers. In this case, the value ob‐jects in the reducer’s iterator are reused, so you need to copy any thatyou need to retain between calls to the iterator (see Example 9-11).Finally, note that the Mapper’s run() method is public and may be customized by users.MultithreadedMapper is an implementation that runs mappers concurrently in a con‐figurable number of threads (set by mapreduce.mapper.multithreadedmapper.threads).
For most data processing tasks, it confers no advantage over the defaultimplementation. However, for mappers that spend a long time processing each record—because they contact external servers, for example—it allows multiple mappers to runin one JVM with little contention.FileInputFormatFileInputFormat is the base class for all implementations of InputFormat that use filesas their data source (see Figure 8-2).
It provides two things: a place to define which filesare included as the input to a job, and an implementation for generating splits for theinput files. The job of dividing splits into records is performed by subclasses.222| Chapter 8: MapReduce Types and FormatsFigure 8-2. InputFormat class hierarchyFileInputFormat input pathsThe input to a job is specified as a collection of paths, which offers great flexibility inconstraining the input. FileInputFormat offers four static convenience methods forsetting a Job’s input paths:publicpublicpublicpublicstaticstaticstaticstaticvoidvoidvoidvoidaddInputPath(Job job, Path path)addInputPaths(Job job, String commaSeparatedPaths)setInputPaths(Job job, Path...
inputPaths)setInputPaths(Job job, String commaSeparatedPaths)The addInputPath() and addInputPaths() methods add a path or paths to the list ofinputs. You can call these methods repeatedly to build the list of paths. The setInputPaths() methods set the entire list of paths in one go (replacing any paths set on theJob in previous calls).A path may represent a file, a directory, or, by using a glob, a collection of files anddirectories. A path representing a directory includes all the files in the directory as inputto the job.
See “File patterns” on page 66 for more on using globs.Input Formats|223The contents of a directory specified as an input path are not pro‐cessed recursively. In fact, the directory should only contain files. Ifthe directory contains a subdirectory, it will be interpreted as a file,which will cause an error. The way to handle this case is to use a fileglob or a filter to select only the files in the directory based on a namepattern. Alternatively, mapreduce.input.fileinputformat.input.dir.recursive can be set to true to force the input directoryto be read recursively.The add and set methods allow files to be specified by inclusion only. To exclude certainfiles from the input, you can set a filter using the setInputPathFilter() method onFileInputFormat.
Filters are discussed in more detail in “PathFilter” on page 67.Even if you don’t set a filter, FileInputFormat uses a default filter that excludes hiddenfiles (those whose names begin with a dot or an underscore). If you set a filter by callingsetInputPathFilter(), it acts in addition to the default filter. In other words, onlynonhidden files that are accepted by your filter get through.Paths and filters can be set through configuration properties, too (Table 8-4), which canbe handy for Streaming jobs. Setting paths is done with the -input option for theStreaming interface, so setting paths directly usually is not needed.Table 8-4.
Input path and filter propertiesProperty nameTypeDefault value Descriptionmapreduce.input.fileinputformat.inputdirComma-separated paths Nonemapreduce.input.pathFilter.classPathFilterNoneThe input files for a job. Paths that containcommas should have those commas escaped by abackslash character. For example, the glob{a,b} would be escaped as {a\,b}.The filter to apply to the input files for a job.classnameFileInputFormat input splitsGiven a set of files, how does FileInputFormat turn them into splits? FileInputFormat splits only large files—here, “large” means larger than an HDFS block. The split sizeis normally the size of an HDFS block, which is appropriate for most applications;however, it is possible to control this value by setting various Hadoop properties, asshown in Table 8-5.224|Chapter 8: MapReduce Types and FormatsTable 8-5.
![](https://s.studizba.com/z.php?f=/templates/si/i/noavatar.png&w=100&h=100&t=1"){kind=avatar}
