pig-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Apache Wiki <wikidi...@apache.org>
Subject [Pig Wiki] Update of "PigSkewedJoinSpec" by SriranjanManjunath
Date Thu, 07 May 2009 20:55:34 GMT
Dear Wiki user,

You have subscribed to a wiki page or wiki category on "Pig Wiki" for change notification.

The following page has been changed by SriranjanManjunath:
http://wiki.apache.org/pig/PigSkewedJoinSpec

------------------------------------------------------------------------------
  [[Anchor(Use_cases)]]
  == Use cases ==
  
- Skewed join can be used when the underlying data is sufficiently skewed and the user needs
a finer control over the allocation of reducers to counteract the skew. It should also be
used when the tables are too large to fit in memory.
+ Skewed join can be used when the underlying data is sufficiently skewed and the user needs
a finer control over the allocation of reducers to counteract the skew. It should also be
used when the data associated with a given key is too large to fit in memory.
  
  {{{
  big = LOAD 'big_data' AS (b1,b2,b3);
@@ -26, +26 @@

  [[Anchor(Implementation)]]
  == Implementation ==
  
- Skewed join translates into two map/reduce jobs - Sample and Join. The first job samples
the input records and computes a histogram of the underlying key space. The second map/reduce
job partitions the input table and performs a join on the predicate. In order to join the
two tables, one of the tables is partitioned and other is streamed to the map tasks. The map
task of this job uses the ~-pig.quantiles-~ file to determine the number of reducers per key.
It then sends the key to each of the reducers in a round robin fashion. Skewed joins happen
in the reduce phase. 
+ Skewed join translates into two map/reduce jobs - Sample and Join. The first job samples
the input records and computes a histogram of the underlying key space. The second map/reduce
job partitions the input table and performs a join on the predicate. In order to join the
two tables, one of the tables is partitioned and other is streamed to the reducer. The map
task of this job uses the ~-pig.quantiles-~ file to determine the number of reducers per key.
It then sends the key to each of the reducers in a round robin fashion. Skewed joins happen
in the reduce phase. 
  
  attachment:partition.jpg
  
@@ -35, +35 @@

  If the underlying data is sufficiently skewed, load imbalances will result in a few reducers
getting a lot of keys. As a first task, the sampler creates a histogram of the key distribution
and stores it in the ~-pig.keydist-~ file. This key distribution will be used to allocate
the right number of reducers for a key. For the table which is partitioned, the partitioner
uses the key distribution to send the data to the reducer in a round robin fashion. For the
table which is streamed, the mapper task uses the ~-pig.keydist-~ file to copy the data to
each of the reduce partitions. 
  
  As a first stab at the implementation, we will be using the uniform random sampler used
by Order BY. The sampler currently does not output the key distribution nor the size of the
sample record. It will be modified to support the same.
+ 
- [[Anchor(Sort_phase)]]
- === Sort phase ===
- The keys are sorted based on the input predicate.
  [[Anchor(Join_phase)]]
  === Join Phase ===
  Skewed join happens in the reduce phase. As a convention, the first table in the join command
is partitioned and sent to the various reducers. Partitioning allows us to support massive
tables without having to worry about the memory limitations. The partitioner is overridden
to send the data in a round robin fashion to each of the reducers associated with a key. The
partitioner obtains the reducer information from the key distribution file. To counteract
the issues with reducer starvation (i.e. the keys that require more than 1 reducer are granted
the reducers whereas the other keys are starved for the reducers), the user is allowed to
set a config parameter pig.mapreduce.skewedjoin.uniqreducers. The value is a percentage of
unique reducers the partitioner should use. For ex: if the value is 90, 10% of the total reducers
will be used for highly skewed data. If the input is highly skewed and the number of reducers
is very low, the task will bail out and report an error.

Mime
View raw message