Hadoop: Pig Query Optimization
Hadoop : Pig Query Optimization 
Before discussing, how to optimize Pig Latin, it is worth understanding what items tend to create bottlenecks in Pig jobs: 
   
   Input size: It does not seem that a massively parallel system should be I/O bound. Hadoop’s parallelism reduces I/O bound but does not entirely remove it.  
   Shuffle size: The data that is moved from your map tasks to your reduce tasks. All of this data has to be serialized, sorted, moved over the network, merged, and deserialized. Also, the number of maps and reduces matters. So if there are m maps and r reduces, the shuffle will have m x r network connections.  
   Output size: Every record written out by a MapReduce job has to be serialized, possibly compressed, and written to the store.  
   Intermediate results size: The size of these intermediate results is affected by the input size and output size factors mentioned previously. 
   Memory: Some calculations require your job to hold a lot of information in memory, for example, joins. If Pig cannot hold all of the values in memory simultaneously, it will need to spill some to disk. This causes a significant slowdown, as records must be written to and read from disk, possibly multiple times. 
   
  There are a number of things you can do when writing Pig Latin scripts to help reduce the bottlenecks discussed earlier: 
   
   Use Right Data Type during loading the data: If types are not specified in the load statement, Pig assumes the type of =double= for numeric computations. A lot of the time, your data would be much smaller, maybe, integer or long. Specifying the real type will help with a speed of arithmetic computation. It has an additional advantage of early error detection. 
   
  
A = load '<yourfile location>' as (a, b, c);
B = foreach A generate a + b;


A = load '<yourfile location>' as (a: int, a: int, c);
B = foreach A generate a + b;
 
  The Query 2 will run more efficiently than Query 1. In some of our queries with see 2x speedup. 
   
   Set Up Your Joins Properly: Joins are one of the most common data operations, and also one of the costliest. Choosing the correct join implementation can improve your performance significantly. 
   
   
   
   
   
   Project Early and Often:  
   
  
A = load '<yourfile location>' as (a, b, c);
B = load '<yourfile location>' as (x, y, z);
C = join A by a, B by x;
D = group C by b;
E = foreach D generate group, COUNT($1);
 
  There is no need for c, y, or z to participate in this query.And there is no need to carry both a and x past the join, just one will suffice.Change the above query- 
  --Query 2
A = load '<yourfile location>' as (a, b, c);
A1 = foreach A generate a, b;
B = load '<yourfile location>' as (x, y, z);
B1 = foreach B generate x;
C = join A1 by a, B1 by x;
C1 = foreach C generate a, b;
D = group C1 by b;
E = foreach D generate group, COUNT($1);
 
  The Query 2 will run more efficiently than Query 1. In some of our queries with see total time drop by 50%. 
   
   Filter Early and Often:  
   
  
A = load '<yourfile location>' as (a, b, c);
B = load '<yourfile location>' as (x, y, z);
C = join A by a, B by x;
D = group C by b;
E = foreach D generate group, COUNT($1);
F = filter E by C.a == 1;
 
  In the above, query has filter command but it can use early to optimize the query, as a very large amount of data going into the join. 
  
A = load '<yourfile location>' as (a, b, c);
B = load '<yourfile location>' as (x, y, z);
C = filter A by a == 1;
D = join C by a, B by x;
E = group D by b;
F = foreach E generate group, COUNT($1);
 
  The Query 2 will run more efficiently than Query 1. Because it reduces the amount of data going into the join. 
   
   Drop Null Before a join: 
   
  A = load '<yourfile location>' as (a, b, c);
B = load '<yourfile location>' as (x, y, z);
A1 = filter A by a is not null;
B1 = filter B by x is not null;
C = join A1 by a, B1 by x;
 
  The nulls will be dropped before the join. Since all null keys go to a single reducer, if your key is null even a small percentage of the time the gain can be significant. In one test where the key was null 7% of the time and the data was spread across 200 reducers, we saw about a 10x speed up in the query by adding the early filters. 
   
   Use the LIMIT Operator: Often you are not interested in the entire output but rather a sample or top results 
   
  A = load '<yourfile location>' as (a, b, c);
B = order A by a;
C = limit B 500;
 
   
   Using Compression in Intermediate Results: Compression can be used to reduce the amount of data to be stored to disk and written over the network. By default, compression is turned off, both between map and reduce tasks and between MapReduce jobs. 
   
  Thanks...!!
 
