Getting rid of data as quickly as possible will help your script perform better. Pushing filters higher in your script may reduce the amount of data you are shuffling or storing in HDFS between MapReduce jobs. Pig's logical optimizer will push your filters up whenever it can. In cases where a filter has multiple predicates joined by and and one or more of the predicates can be applied before the operator preceding the filter, Pig will split the filter at the and and push the eligible predicate(s). This allows Pig to push parts of the filter when it may not be able to push the filter as a whole. Table 8.1, “When Pig Pushes Filters” describes when these filter predicates will and will not be pushed once they have been split.

Also, consider adding additional filters that are implicit in your script. For example, all of the records with null values in the key will be thrown out by an inner join. If you know that more than a few hundred of your records have null key values, put a filter input by key is not null before the join. This will enhance the performance of your join.

We used to tell users to use foreach to remove fields they were not using as soon as possible. As of version 0.8 Pig's logical optimizer does a fair job of removing fields aggressively when it can tell that they will no longer be used.

-- itemid does not need to be loaded, since it is not used in the script
txns        = load 'purchases' as (date, storeid, amount, itemid);
todays      = filter txns by date == '20110513'; -- date not needed after this
bystore     = group todays by storeid;
avgperstore = foreach bystore generate group, AVG(todays.amount);

However, you are still smarter than Pig's optimizer, so there are situations where you can tell that a field is no longer needed but Pig cannot. If AVG(todays.amount) were changed to COUNT(todays) in the above example, Pig would not be able to determine that, after the filter, only storeid and amount were required. It cannot see that COUNT does not need all of the fields in the bag it is being passed. Whenever you pass a UDF the entire record (udf(*)) or an entire complex field Pig cannot determine which fields are required. In this case you will need to put in the foreach yourself to remove unneeded data as early as possible.

Joins are one of the most common data operations, and also one of the costliest. Choosing the correct join implementation can significantly improve your performance. The flowchart in Figure 8.1, “Choosing a Join Implementation” will help you select the correct join implementation.

Figure 8.1. Choosing a Join Implementation

Once you have selected your join implementation make sure to arrange your inputs in the correct order as well. For replicated joins, the small table must be given as the last input. For skewed joins, the second input is the one that is sampled for large keys. For the default join the right-most input has its records streamed through, while the other input(s) have their records for a given key value materialized in memory. Thus if you have one join input that you know has more records per key value, you should place it right-most in the join. For merge join the left input is taken as the input for the MapReduce job, and thus the number of maps started are based on this input. If one input is much larger than the other you should place it on the left in order to get more map tasks dedicated to your jobs. This will also reduce the size of the sampling step that builds the index for the right side. For complete details on each of these join implementations see the section called “Join” and the section called “Using Different Join Implementations”.

Whenever you are doing operations that can be combined by multi-query such as grouping and filtering, these should be written together in one Pig Latin script so that Pig can combine them. While adding additional operations does add some time to the total processing time, it is still much faster than running jobs separately.

As has been discussed elsewhere, Pig can run with or without data type information. In cases where the load function you are using creates already typed data, there is little you need to do to optimize the performance. However, if you are using the default PigStorage load function that reads tab delimited files, then whether you use types will affect your performance.

On the one hand, converting fields from bytearray to the appropriate type has a cost. So, if you do not need type information, you should not declare it. For example, if you are just counting records, you can omit the type declaration and not affect the outcome of your script.

On the other hand, if you are doing integer calculcations, types can help your script perform better. When Pig is asked to do a numeric calculation on a bytearray it treats that bytearray as a double, since this is the safest assumption. But floating point arithmetic is much slower than integer arithmetic on most machines. For example, if you are doing a SUM over integer values, you will get better performance by declaring them to be of type integer.

Setting your parallelism properly can be difficult, as there are a number of factors. Before we discuss the factors, a little background will be helpful. It would be natural to think more parallelism is always better. That is not the case. Like any other resource, parallelism has a network cost as discussed under performance bottlenecks in the shuffle phase at the beginning of this chapter.

The second way increasing parallelism adds latency to your script is that there is a limited number of reduce slots in your cluster, or a limited number that your scheduler will assign to you. If 100 reduce slots are available to you and you specify parallel 200 you will still only be able to run 100 reduces at a time. Your reducers will run in two separate waves. Since there is overhead in starting and stopping reduce tasks and the shuffle gets less efficient as parallelism increases, it is often not efficient to select more reducers than you have slots to run them. In fact it is best to specify slightly fewer reducers than slots that you can access. This leaves room for MapReduce to restart a few failed reducers and to use speculative execution without doubling your reduce time. See the section called “Handling Failure” for information on speculative execution.

Also, it is important to keep in mind the affects of skew on parallelism. MapReduce generally does a good job partitioning keys equally to the reducers. But the number of records per key often varies radically. Thus a few reducers that get keys with a large number of records will significantly lag the other reducers. Pig cannot start the next MapReduce job until all of the reducers have finished in the previous job. So the slowest reducer defines the length of the job. If you have 10G of input to your reducers and you set parallel to 10, but one key accounts for 50% of the data (a not uncommon case), then nine of your reducers will finish quite quickly while the last lags. Increasing your parallelism will not help, it will just waste more cluster resources. Instead you need to use Pig's mechanisms to handle skew.