Push limit into loader

Details

Description

We can optimize limit operation by stopping early in PigRecordReader. In general, we need a way to communicate between PigRecordReader and execution pipeline. POLimit could instruct PigRecordReader that we have already had enough records and stop feeding more data.

Since with your patch, the LimitOptimizer would remove LOLimit from logic plans after set the limit to LOLoad, this would generate a map-only job. Record number of the result would be map_num * 10, this is incorrect.

Min Zhou
added a comment - 24/Nov/11 07:34 @Daniel
It improves here, but with a bug. I did the test in a 25-nodes cluster which such script
A = load '/tpch/orders' USING PigStorage('\u0001') AS (o_orderkey:int, o_custkey:int, o_orderstatus:chararray, o_totalprice:double, o_orderdate:chararray, o_orderpriority:chararray, o_clerk:chararray, o_shippriority:int, o_comment: chararray);
F = FOREACH A GENERATE o_orderkey;
L = LIMIT F 10;
DUMP L;
case
job cost time
HDFS bytes read
Average time taken by Map tasks
Worst performing map task
w/o optimization
26 sec
12,976,128
1 sec
1 sec
with optimization
24 sec
19,347,931,305
3 sec
5 sec
Since with your patch, the LimitOptimizer would remove LOLimit from logic plans after set the limit to LOLoad, this would generate a map-only job. Record number of the result would be map_num * 10, this is incorrect.
I will submit a patch soon.

Daniel Dai
added a comment - 28/Nov/11 01:49 Thanks Min, that's encouraging. Which version of Hadoop are you using?
Also I discussed with Min in IM, it will be better to have a global flag to signal the sufficiency of records, so that we can address more cases for this optimization.

We are using a modified version of 0.19.1. However, that internal version provide new MR API and is compatible with both hadoop clients under the versions of 0.19.x and 0.20.2. Our version doesn't change any logic of map phase from the community version, so this patch should improves the latter as well.

That's a good attempt if we can address more cases like limit optimization on LOFilter.

Min Zhou
added a comment - 30/Nov/11 04:46 We are using a modified version of 0.19.1. However, that internal version provide new MR API and is compatible with both hadoop clients under the versions of 0.19.x and 0.20.2. Our version doesn't change any logic of map phase from the community version, so this patch should improves the latter as well.
That's a good attempt if we can address more cases like limit optimization on LOFilter.

Daniel in your original comment the script mentioned is similar to a "SELECT * .. LIMIT 10" Hive currently does not run a M/R job for these situations, it just reads the data and streams it to stdout. Can we do such an optimization for the query mentioned?

Additionally can we use some optimizations that Hadoop 23 has such as running in an Uberized task rather than launch M/R jobs?

Viraj Bhat
added a comment - 10/Feb/12 03:33 What version is this likely going to be fixed?
Daniel in your original comment the script mentioned is similar to a "SELECT * .. LIMIT 10" Hive currently does not run a M/R job for these situations, it just reads the data and streams it to stdout. Can we do such an optimization for the query mentioned?
Additionally can we use some optimizations that Hadoop 23 has such as running in an Uberized task rather than launch M/R jobs?
Viraj

@Viraj
For the patch itself, I would like to commit it into trunk soon. For the direct hdfs access you mention, it will probably be part of the backend rework we planned, but I am not sure at the moment.

Daniel Dai
added a comment - 13/Feb/12 08:23 @Viraj
For the patch itself, I would like to commit it into trunk soon. For the direct hdfs access you mention, it will probably be part of the backend rework we planned, but I am not sure at the moment.