Tom White - Hadoop The Definitive Guide_ 4 edition - 2015 (811394), страница 56
Текст из файла (страница 56)
IntPair’stoString() method creates a tab-separated string, so the output is a set of tab-separatedyear-temperature pairs.Many applications need to access all the sorted values, not just thefirst value as we have provided here. To do this, you need to popu‐late the value fields since in the reducer you can retrieve only the firstkey. This necessitates some unavoidable duplication of informationbetween key and value.We set the partitioner to partition by the first field of the key (the year) using a custompartitioner called FirstPartitioner. To sort keys by year (ascending) and temperature(descending), we use a custom sort comparator, using setSortComparatorClass(), thatextracts the fields and performs the appropriate comparisons.
Similarly, to group keysby year, we set a custom comparator, using setGroupingComparatorClass(), to extractthe first field of the key for comparison.5Running this program gives the maximum temperatures for each year:% hadoop jar hadoop-examples.jar MaxTemperatureUsingSecondarySort \input/ncdc/all output-secondarysort% hadoop fs -cat output-secondarysort/part-* | sort | head1901 3171902 2441903 2891904 2561905 2831906 2941907 2831908 2891909 2781910 294StreamingTo do a secondary sort in Streaming, we can take advantage of a couple of library classesthat Hadoop provides. Here’s the driver that we can use to do a secondary sort:% hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-*.jar \-D stream.num.map.output.key.fields=2 \-D mapreduce.partition.keypartitioner.options=-k1,1 \-D mapreduce.job.output.key.comparator.class=\5.
For simplicity, these custom comparators as shown are not optimized; see “Implementing a RawComparatorfor speed” on page 123 for the steps we would need to take to make them faster.266|Chapter 9: MapReduce Featuresorg.apache.hadoop.mapred.lib.KeyFieldBasedComparator \-D mapreduce.partition.keycomparator.options="-k1n -k2nr" \-files secondary_sort_map.py,secondary_sort_reduce.py \-input input/ncdc/all \-output output-secondarysort-streaming \-mapper ch09-mr-features/src/main/python/secondary_sort_map.py \-partitioner org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner \-reducer ch09-mr-features/src/main/python/secondary_sort_reduce.pyOur map function (Example 9-7) emits records with year and temperature fields.
Wewant to treat the combination of both of these fields as the key, so we setstream.num.map.output.key.fields to 2. This means that values will be empty, justlike in the Java case.Example 9-7. Map function for secondary sort in Python#!/usr/bin/env pythonimport reimport sysfor line in sys.stdin:val = line.strip()(year, temp, q) = (val[15:19], int(val[87:92]), val[92:93])if temp == 9999:sys.stderr.write("reporter:counter:Temperature,Missing,1\n")elif re.match("[01459]", q):print "%s\t%s" % (year, temp)However, we don’t want to partition by the entire key, so we useKeyFieldBasedPartitioner, which allows us to partition by a part of the key.
Thespecification mapreduce.partition.keypartitioner.options configures the parti‐tioner. The value -k1,1 instructs the partitioner to use only the first field of the key,where fields are assumed to be separated by a string defined by themapreduce.map.output.key.field.separator property (a tab character by default).Next, we want a comparator that sorts the year field in ascending order and the tem‐perature field in descending order, so that the reduce function can simply return thefirst record in each group.
Hadoop provides KeyFieldBasedComparator, which is idealfor this purpose. The comparison order is defined by a specification that is like the oneused for GNU sort. It is set using the mapreduce.partition.keycomparator.optionsproperty. The value -k1n -k2nr used in this example means “sort by the first field innumerical order, then by the second field in reverse numerical order.” Like its partitionercousin, KeyFieldBasedPartitioner, it uses the map output key separator to split a keyinto fields.In the Java version, we had to set the grouping comparator; however, in Streaming,groups are not demarcated in any way, so in the reduce function we have to detect thegroup boundaries ourselves by looking for when the year changes (Example 9-8).Sorting|267Example 9-8.
Reduce function for secondary sort in Python#!/usr/bin/env pythonimport syslast_group = Nonefor line in sys.stdin:val = line.strip()(year, temp) = val.split("\t")group = yearif last_group != group:print vallast_group = groupWhen we run the Streaming program, we get the same output as the Java version.Finally, note that KeyFieldBasedPartitioner and KeyFieldBasedComparator are notconfined to use in Streaming programs; they are applicable to Java MapReduce pro‐grams, too.JoinsMapReduce can perform joins between large datasets, but writing the code to do joinsfrom scratch is fairly involved. Rather than writing MapReduce programs, you mightconsider using a higher-level framework such as Pig, Hive, Cascading, Cruc, or Spark,in which join operations are a core part of the implementation.Let’s briefly consider the problem we are trying to solve.
We have two datasets—forexample, the weather stations database and the weather records—and we want to rec‐oncile the two. Let’s say we want to see each station’s history, with the station’s metadatainlined in each output row. This is illustrated in Figure 9-2.How we implement the join depends on how large the datasets are and how they arepartitioned. If one dataset is large (the weather records) but the other one is small enoughto be distributed to each node in the cluster (as the station metadata is), the join can beeffected by a MapReduce job that brings the records for each station together (a partialsort on station ID, for example). The mapper or reducer uses the smaller dataset to lookup the station metadata for a station ID, so it can be written out with each record.
See“Side Data Distribution” on page 273 for a discussion of this approach, where we focus onthe mechanics of distributing the data to nodes in the cluster.268|Chapter 9: MapReduce FeaturesFigure 9-2. Inner join of two datasetsIf the join is performed by the mapper it is called a map-side join, whereas if it is per‐formed by the reducer it is called a reduce-side join.If both datasets are too large for either to be copied to each node in the cluster, we canstill join them using MapReduce with a map-side or reduce-side join, depending onhow the data is structured. One common example of this case is a user database and alog of some user activity (such as access logs).
For a popular service, it is not feasible todistribute the user database (or the logs) to all the MapReduce nodes.Map-Side JoinsA map-side join between large inputs works by performing the join before the datareaches the map function. For this to work, though, the inputs to each map must bepartitioned and sorted in a particular way. Each input dataset must be divided into theJoins|269same number of partitions, and it must be sorted by the same key (the join key) in eachsource. All the records for a particular key must reside in the same partition.
This maysound like a strict requirement (and it is), but it actually fits the description of the outputof a MapReduce job.A map-side join can be used to join the outputs of several jobs that had the same numberof reducers, the same keys, and output files that are not splittable (by virtue of beingsmaller than an HDFS block or being gzip compressed, for example). In the context ofthe weather example, if we ran a partial sort on the stations file by station ID, and anotheridentical sort on the records, again by station ID and with the same number of reducers,then the two outputs would satisfy the conditions for running a map-side join.You use a CompositeInputFormat from the org.apache.hadoop.mapreduce.joinpackage to run a map-side join.
The input sources and join type (inner or outer) forCompositeInputFormat are configured through a join expression that is written ac‐cording to a simple grammar. The package documentation has details and examples.The org.apache.hadoop.examples.Join example is a general-purpose command-lineprogram for running a map-side join, since it allows you to run a MapReduce job forany specified mapper and reducer over multiple inputs that are joined with a given joinoperation.Reduce-Side JoinsA reduce-side join is more general than a map-side join, in that the input datasets don’thave to be structured in any particular way, but it is less efficient because both datasetshave to go through the MapReduce shuffle.
The basic idea is that the mapper tags eachrecord with its source and uses the join key as the map output key, so that the recordswith the same key are brought together in the reducer. We use several ingredients tomake this work in practice:Multiple inputsThe input sources for the datasets generally have different formats, so it is veryconvenient to use the MultipleInputs class (see “Multiple Inputs” on page 237) toseparate the logic for parsing and tagging each source.Secondary sortAs described, the reducer will see the records from both sources that have the samekey, but they are not guaranteed to be in any particular order.
![](https://s.studizba.com/z.php?f=/templates/si/i/noavatar.png&w=100&h=100&t=1"){kind=avatar}
