您可能需要检查以下几项：-
1> 过滤掉f1和f2值都为空的记录（如果有）
2> 如果可能，尝试通过实现代数接口来使用hadoop combiner：-
https://www.safaribooksonline.com/library/view/programming-pig/9781449317881/ch10s02.html
3> 使用CustomPartitioner使用另一个键跨reducer分发数据。
下面是我用来在join之后对倾斜数据进行分区的示例代码（同样也可以在group之后使用）：-

import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.io.NullableTuple;
import org.apache.pig.impl.io.PigNullableWritable;

public class KeyPartitioner extends Partitioner<PigNullableWritable, Writable> {

/**

* Here key contains value of current key used for partitioning and Writable
* value conatins all fields from your tuple. I used my 5th field from tuple to do partitioning as I knew it has evenly distributed value.
**/

@Override
public int getPartition(PigNullableWritable key, Writable value, int numPartitions) {
    Tuple valueTuple = (Tuple) ((NullableTuple) value).getValueAsPigType();
    try {
        if (valueTuple.size() > 5) {
            Object hashObj = valueTuple.get(5);
            Integer keyHash = Integer.parseInt(hashObj.toString());
            int partitionNo = Math.abs(keyHash) % numPartitions;
            return partitionNo;
        } else {
            if (valueTuple.size() > 0) {
                return (Math.abs(valueTuple.get(1).hashCode())) % numPartitions;
            }
        }
    } catch (NumberFormatException | ExecException ex) {
        Logger.getLogger(KeyPartitioner.class.getName()).log(Level.SEVERE, null, ex);
    }
    return (Math.abs(key.hashCode())) % numPartitions;
}
}