Pig Mix

PigMix is a set of queries used test pig performance from release to release. There are queries that test latency (how long does it take to run this query?), and queries that test scalability (how many fields or records can pig handle before it fails?). In addition it includes a set of map reduce java programs to run equivalent map reduce jobs directly. These will be used to test the performance gap between direct use of map reduce and using pig.

Runs

The following table includes runs done of the pig mix. All of these runs have been done on a cluster with 26 slaves plus one machine acting as the name node and job tracker. The cluster was running hadoop version 0.18.1. (TODO: Need to get specific hardware info on those machines).

The tests were run against two versions of pig: top of trunk, and top of types branch both as of Nov 21 2008.

The tests were run three times for each version and the results averaged.

tot = top of trunk totb = top of types branch

The totb run of 1/20/09 includes the change to make BufferedPositionedInputStream use a buffer instead of relying on hadoop to buffer.

tot run of 2/23/09, top of trunk is now what was on the types branch (that is proto 0.2.0). This run includes fragment replicate join and rework of partitioning for order by.

Run of 5/28/09, placed in a separate table because there were underlying cluster changes, thus the map reduce tests needed to be rerun. This is the same code base that became 0.3.0.

|| Version || Map Reduce Java code || tot 5/27/09 ||

Run date: June 28, 2009, run against top of trunk as of that day. Note that the columns got reversed in this one (Pig then MR)

|| Test || Pig run time || Java run time || Multiplier ||

Run date: August 27, 2009, run against top of trunk as of that day.

Run date: October 18, 2009, run against top of trunk as of that day. With this run we included a new measure, weighted average. Our previous multiplier that we have been publishing takes the total time of running all 12 Pig Latin scripts and compares it to the total time of running all 12 Java Map Reduce programs. This is a valid way to measure, as it shows the total amount of time to do all these operations on both platforms. But it has the drawback that it gives more weight to long running operations (such as joins and order bys) while masking the performance in faster operations such as group bys. The new "weighted average" adds up the multiplier for each Pig Latin script vs. Java program separately and then divides by 12, thus weighting each test equally. In past runs the weighted average had significantly lagged the overall average (for example, in the run above for August 27 it was 1.5 even though the total difference was 1.2). With this latest run it still lags some, but the gap has shrunk noticably.

Features Tested

Based on a sample of user queries, PigMix includes tests for the following features.

1. Data with many fields, but only a few are used.
2. Reading data from maps.
3. Use of bincond and arithmetic operators.
4. Exploding nested data.
5. Load bzip2 data
6. Load uncompressed data
7. join with one table small enough to fit into a fragment and replicate algorithm.
8. join where tables are sorted and partitioned on the same key
9. Do a cogroup that is not immediately followed by a flatten (that is, use cogroup for something other than a straight forward join).
10. group by with only algebraic udfs that has nested plan (distinct aggs basically).
11. foreachs with nested plans including filter and implicit splits.
12. group by where the key accounts for a large portion of the record.
13. group all
14. union plus distinct
15. order by
16. multi-store query (that is, a query where data is scanned once, then split and grouped different ways).

The data is generated so that it has a zipf type distribution for the group by and join keys, as this models most human generated data. Some other fields are generated using a uniform data distribution.

Scalability tests test the following:

1. Join of very large data sets.
2. Grouping of very large data set.
3. Query with a very wide (500+ fields) row.
4. Loading many data sets together in one load

Proposed Data

Initially, four data sets have been created. The first, page_views, is 10 million rows in size, with a schema of:

The second, users, was created by taking the unique user keys from page_views and adding additional columns.

The third, power_users, has 500 rows, and has the same schema as users. It was generated by skimming 500 unique names from users. This will produce a table that can be used to test fragment replicate type joins.

The fourth, widerow, has a very wide row (500 fields), consisting of one string and 499 integers.

users, power_users, and widerow are written in ASCII format, using Ctrl-A as the field delimiter. They can be read using PigStorage.

page_views is written in as text data, with Ctrl-A as the field delimiter. Maps in the file are delimited by Ctrl-C between key value pairs and Ctrl-D between keys and values. Bags in the file are delimited by Ctrl-B between tuples in the bag. A special loader, PigPerformance loader has been written to read this format.

Proposed Scripts

Scalability

Script S1

This script tests grouping, projecting, udf envocation, and filtering with a very wide row. Covers scalability feature 3.

A = load '$widerow' using PigStorage('\u0001') as (name: chararray, c0: int, c1: int, ..., c500: int);
B = group A by name parallel $parrallelfactor;
C = foreach B generate group, SUM(A.c0) as c0, SUM(A.c1) as c1, ... SUM(A.c500) as c500;
D = filter C by c0 > 100 and c1 > 100 and c2 > 100 ... and c500 > 100;
store D into '$out';

Script S2 This script tests joining two inputs where a given value of the join key appears many times in both inputs. This will test pig's ability to handle large joins. It covers scalability features 1 and 2.

TBD

Features not yet tested: 4.

Latency

Script L1

This script tests reading from a map, flattening a bag of maps, and use of bincond (features 2, 3, and 4).

register pigperf.jar;
A = load '$page_views' using org.apache.pig.test.utils.datagen.PigPerformanceLoader()
    as (user, action, timespent, query_term, ip_addr, timestamp,
        estimated_revenue, page_info, page_links);
B = foreach A generate user, (int)action as action, (map[])page_info as page_info,
    flatten((bag{tuple(map[])})page_links) as page_links;
C = foreach B generate user,
    (action == 1 ? page_info#'a' : page_links#'b') as header;
D = group C by user $parallelfactor;
E = foreach D generate group, COUNT(C) as cnt;
store E into '$out';

Script L2

This script tests using a join small enough to do in fragment and replicate (feature 7).

register pigperf.jar;
A = load '$page_views' using org.apache.pig.test.utils.datagen.PigPerformanceLoader()
    as (user, action, timespent, query_term, ip_addr, timestamp,
        estimated_revenue, page_info, page_links);
B = foreach A generate user, estimated_revenue;
alpha = load '$power_users' using PigStorage('\u0001') as (name, phone,
        address, city, state, zip);
beta = foreach alpha generate name;
C = join beta by name, A by user $parallelfactor;
store C into '$out';

Script L3

This script tests a join too large for fragment and replicate. It also contains a join followed by a group by on the same key, something that pig could potentially optimize by not regrouping.

register pigperf.jar;
A = load '$page_views' using org.apache.pig.test.utils.datagen.PigPerformanceLoader()
    as (user, action, timespent, query_term, ip_addr, timestamp,
        estimated_revenue, page_info, page_links);
B = foreach A generate user, (double)estimated_revenue;
alpha = load '$users' using PigStorage('\u0001') as (name, phone, address,
        city, state, zip);
beta = foreach alpha generate name;
C = join beta by name, A by user $parallelfactor;
D = group C by $0 $parallelfactor;
E = foreach D generate group, SUM(C.estimated_revenue);
store E into '$out';

Script L4

This script covers foreach generate with a nested distinct (feature 10).

register pigperf.jar;
A = load '$page_views' using org.apache.pig.test.utils.datagen.PigPerformanceLoader()
    as (user, action, timespent, query_term, ip_addr, timestamp,
        estimated_revenue, page_info, page_links);
B = foreach A generate user, action;
C = group B by user $parallelfactor;
D = foreach C {
    aleph = B.action;
    beth = distinct aleph;
    generate group, COUNT(beth);
}
store D into '$out';

Script L5

This script does an anti-join. This is useful because it is a use of cogroup that is not a regular join (feature 9).

register pigperf.jar;
A = load '$page_views' using org.apache.pig.test.utils.datagen.PigPerformanceLoader()
    as (user, action, timespent, query_term, ip_addr, timestamp,
        estimated_revenue, page_info, page_links);
B = foreach A generate user;
alpha = load '$users' using PigStorage('\u0001') as (name, phone, address,
        city, state, zip);
beta = foreach alpha generate name;
C = cogroup beta by name, A by user $parallelfactor;
D = filter C by COUNT(beta) == 0;
E = foreach D generate group;
store E into '$out';

Script L6

This script covers the case where the group by key is a significant percentage of the row (feature 12).

register pigperf.jar;
A = load '$page_views' using org.apache.pig.test.utils.datagen.PigPerformanceLoader()
    as (user, action, timespent, query_term, ip_addr, timestamp,
        estimated_revenue, page_info, page_links);
B = foreach A generate user, action, (int)timespent as timespent, query_term, ip_addr, timestamp;
C = group B by (user, query_term, ip_addr, timestamp) $parallelfactor;
D = foreach C generate flatten(group), SUM(B.timespent);
store D into '$out';

Script L7

This script covers having a nested plan with splits (feature 11).

register pigperf.jar;
A = load '$page_views' using org.apache.pig.test.utils.datagen.PigPerformanceLoader() as (user, action, timespent, query_term,
            ip_addr, timestamp, estimated_revenue, page_info, page_links);
B = foreach A generate user, timestamp;
C = group B by user $parallelfactor;
D = foreach C {
    morning = filter B by timestamp < 43200;
    afternoon = filter B by timestamp >= 43200;
    generate group, COUNT(morning), COUNT(afternoon);
}
store D into '$out';

Script L8

This script covers group all (feature 13).

register pigperf.jar;
A = load '$page_views' using org.apache.pig.test.utils.datagen.PigPerformanceLoader()
    as (user, action, timespent, query_term, ip_addr, timestamp,
        estimated_revenue, page_info, page_links);
B = foreach A generate user, (int)timespent as timespent, (double)estimated_revenue as estimated_revenue;
C = group B all;
D = foreach C generate SUM(B.timespent), AVG(B.estimated_revenue);
store D into '$out';

Script L9

This script covers order by of a single value (feature 15).

register pigperf.jar;
A = load '$page_views' using org.apache.pig.test.utils.datagen.PigPerformanceLoader()
    as (user, action, timespent, query_term, ip_addr, timestamp,
        estimated_revenue, page_info, page_links);
B = order A by query_term $parallelfactor;
store B into '$out';

Script L10

This script covers order by of multiple values (feature 15).

register pigperf.jar;
A = load '$page_views' using org.apache.pig.test.utils.datagen.PigPerformanceLoader()
    as (user, action, timespent:int, query_term, ip_addr, timestamp,
        estimated_revenue:double, page_info, page_links);
B = order A by query_term, estimated_revenue desc, timespent $parallelfactor;
store B into '$out';

Script L11

This script covers distinct and union and reading from a wide row but using only one field (features: 1, 14).

register pigperf.jar;
A = load '$page_views' using org.apache.pig.test.utils.datagen.PigPerformanceLoader()
    as (user, action, timespent, query_term, ip_addr, timestamp,
        estimated_revenue, page_info, page_links);
B = foreach A generate user;
C = distinct B $parallelfactor;
alpha = load '$widerow' using PigStorage('\u0001');
beta = foreach alpha generate $0 as name;
gamma = distinct beta $parallelfactor;
D = union C, gamma;
E = distinct D $parallelfactor;
store E into '$out';

Script L12

This script covers multi-store queries (feature 16).

register pigperf.jar;
A = load '$page_views' using org.apache.pig.test.utils.datagen.PigPerformanceLoader()
    as (user, action, timespent, query_term, ip_addr, timestamp,
        estimated_revenue, page_info, page_links);
B = foreach A generate user, action, (int)timespent as timespent, query_term,
    (double)estimated_revenue as estimated_revenue;
split B into C if user is not null, alpha if user is null;
split C into D if query_term is not null, aleph if query_term is null;
E = group D by user $parallelfactor;
F = foreach E generate group, MAX(D.estimated_revenue);
store F into 'highest_value_page_per_user';
beta = group alpha by query_term $parallelfactor;
gamma = foreach beta generate group, SUM(alpha.timespent);
store gamma into 'total_timespent_per_term';
beth = group aleph by action $parallelfactor;
gimel = foreach beth generate group, COUNT(aleph);
store gimel into 'queries_per_action';

Features not yet covered: 5 (bzip data), 8 (sorted join)

Data Generation

If you want to run this queires yourself, please, see https://issues.apache.org/jira/browse/PIG-200 on how to generate the data. See DataGeneratorHadoop for information on how to run data generator in hadoop mode.