I have a series of Pig scripts that are transforming hundreds of millions of records from multiple data sources that need to be joined together. Towards the end of each script, I reach a point where JOIN performance becomes terribly slow. Looking at the DAG in the Tez View, I see that it is split into relatively few tasks (typically 100-200), but each task takes multiple hours to complete. The task description shows that it's doing a HASH_JOIN.
Interestingly, I only run into this bottleneck when running on the Tez execution engine. On MapReduce, it can still take a while, but nothing like the agonizing crawl I get on Tez. However, running on MapReduce is a problem as I have an issue with MapReduce for which I've asked another question here.
Here's a sample of my code (apologies, I've had to make the code very generic to be able to post on the interwebs). I'm wondering what I can do to remove this bottleneck -- would specifying parallelism help? Is there something wrong with my approach?
-- Incoming data:
-- A: hundreds of millions of rows, 19 fields
-- B: hundreds of millions of rows, 3 fields
-- C: hundreds of millions of rows, 5 fields
-- D: a few thousand rows, 5 fields
J = -- This reduces the size of A, but still probably in the hundreds of millions
FILTER A
BY qualifying == 1;
K = -- This is a one-to-one join that doesn't explode the number of rows in J
JOIN J BY Id
, B BY Id;
L =
FOREACH K
GENERATE J1 AS L1
, J2 AS L2
, J3 AS L3
, J4 AS L4
, J5 AS L5
, J6 AS L6
, J7 AS L7
, J8 AS L8
, B1 AS L9
, B2 AS L10
;
M = -- Reduces the size of C to around one hundred million rows
FILTER C
BY Code matches 'Code-.+';
M_WithYear =
FOREACH M
GENERATE *
, (int)REGEX_EXTRACT(Code, 'Code-.+-([0-9]+)', 1) AS year:int
;
SPLIT M_WithYear
INTO M_annual IF year <= (int)'$currentYear' -- roughly 75% of the data from M
, M_lifetime IF Code == 'Code-Lifetime'; -- roughly 25% of the data from M
-- Transformations for M_annual
N =
JOIN M_WithYear BY Id, D BY Id USING 'replicated';
O = -- This is where performance falls apart
JOIN N BY (Id, year, M7) -- M7 matches L7
, L BY (Id, year, L7);
P =
FOREACH O
GENERATE N1 AS P1
, N2 AS P2
, N3 AS P3
, N4 AS P4
, N5 AS P5
, N6 AS P6
, N7 AS P7
, N8 AS P8
, N9 AS P9
, L1 AS P10
, L2 AS P11
;
-- Transformations N-P above repeated for M_lifetime