Hive - Efficiente unione di due tabelle

Question

sto unendo due grandi tavoli in Hive (uno è oltre 1 miliardo di righe, una è di circa 100 milioni di righe) in questo modo:

create table joinedTable as select t1.id, ... from t1 join t2 ON (t1.id = t2.id); 

ho bucketed le due tabelle in allo stesso modo, raggruppando per ID in 100 bucket per ciascuno, ma la query richiede ancora molto tempo.

Qualche suggerimento su come accelerare?

Answer 1

Come è stato eseguito il bucket dei dati con le chiavi di join, è possibile utilizzare il Bucket Map Join. Per questo la quantità di benne in una tabella deve essere un multiplo della quantità di benne nell'altra tabella. Può essere attivato eseguendo set hive.optimize.bucketmapjoin=true; prima della query. Se i tavoli non soddisfano le condizioni, Hive eseguirà semplicemente il normale Inner Join.

Se entrambe le tabelle hanno la stessa quantità di secchi ei dati vengono ordinati per i tasti della benna, Hive in grado di eseguire più velocemente sort merge join. Per attivarlo, è necessario eseguire i seguenti comandi:

set hive.input.format=org.apache.hadoop.hive.ql.io.BucketizedHiveInputFormat; 
set hive.optimize.bucketmapjoin=true; 
set hive.optimize.bucketmapjoin.sortedmerge=true; 

si possono trovare alcune visualizzazioni dei diversi aderire tecniche sotto https://cwiki.apache.org/confluence/download/attachments/27362054/Hive+Summit+2011-join.pdf.

Answer 2

come la vedo io la risposta è un po 'più complicato di quello che @Adrian Lange offerto.

In primo luogo è necessario capire una differenza molto importante tra il BucketJoin e sort-merge benna registrazione (SMBJ):

per eseguire un bucketjoin "la quantità dei secchi in una tabella deve essere un multiplo della quantità di secchi nell'altra tabella "come indicato in precedenza e inoltre hive.optimize.bucketmapjoin deve essere impostato su true.
emissione di un join, alveare convertirà in un bucketjoin se la condizione di cui sopra si svolgono MA fare attenzione quell'alveare non rispettare la bucket! ciò significa che la creazione della tabella con il bucket non è sufficiente perché la tabella venga effettivamente archiviata nella quantità specificata di bucket, in quanto l'hive non la applica a meno che hive.enforce.bucketing sia impostato su true (il che significa che la quantità di bucket in realtà viene impostata dal quantità di riduttori nella fase finale della query che inserisce i dati nella tabella).
Dal punto di vista delle prestazioni, quando si utilizza un secchio a una singola attività legge la tabella "più piccola" nella cache distribuita prima che i mappatori accedano e facciano il join - Questa fase probabilmente sarebbe molto molto lunga e inefficace quando il tuo tavolo ha ~ 100 milioni di righe!
Dopo l'unione, il join verrà eseguito come in un join regolare eseguito nei riduttori.

Per eseguire un SMBJ entrambe le tabelle devono avere la stessa esatta quantità di benne, sulle stesse colonne e filtrate per queste colonne oltre ad impostare hive.optimize.bucketmapjoin.sortedmerge true.
Come nell'ottimizzazione precedente, Hive non impone il bucket e la selezione, ma piuttosto presuppone che si fatto in modo che le tabelle sono in realtà bucketed e ordinati (non solo per definizione, ma impostando hive.enforce.sorting o l'ordinamento manuale dei dati durante l'inserimento di esso) - Questo è molto importante in quanto potrebbe portare a risultati errati in entrambi i casi.
A partire dal lato performace, questa ottimizzazione è il modo più efficiente per i seguenti motivi:

1. Ogni mapper legge sia secchi e non esiste un unico contesa compito per cache distribuita carico
2. Il join in corso di esecuzione è un merge-sort join in quanto i dati sono già ordinati, il che è molto più efficiente.

Si prega di notare le seguenti considerazioni:

• in entrambi i casi set hive.input.format=org.apache.hadoop.hive.ql.io.BucketizedHiveInputFormat;
  dovrebbe essere eseguito
• in entrambi i casi un /*+ MAPJOIN(b) */ deve essere applicato nella query (subito dopo il select e dove b è il tavolo più piccolo)
• Quanti secchi?
  Questo deve essere visualizzato da questo angolo: la considerazione deve essere applicata rigorosamente alla tabella più grande in quanto ha un impatto maggiore da questa direzione e quest'ultima la configurazione verrà applicata al tavolo più piccolo come un must. Penso che come regola empirica ogni secchio debba contenere da 1 a 3 blocchi, probabilmente da qualche parte vicino a 2 blocchi. quindi, se la dimensione del blocco è 256 MB, mi sembra ragionevole avere ~ 512 MB di dati in ciascun bucket nella tabella più grande, in modo che questo diventi un semplice problema di divisione.

Inoltre, non dimenticare che queste ottimizzazioni da sole non garantiscono sempre un tempo di interrogazione più veloce.
Diciamo che si sceglie di fare un SMBJ, questo aggiunge il costo di ordinare 2 tabelle prima di eseguire il join - quindi più volte si eseguirà la query meno si "paga" per questa fase di ordinamento.

A volte, un join semplice porterà alle migliori prestazioni e nessuna delle ottimizzazioni di cui sopra sarà d'aiuto e sarà necessario ottimizzare il normale processo di join a livello di applicazione/logico o regolando le impostazioni MapReduce/Hive come l'utilizzo della memoria/parallelismo ecc.

Answer 3

Non penso che sia un criterio obbligatorio "la quantità di benne in una tabella deve essere un multiplo della quantità di benne nell'altra tabella" per il join del bucket di mappe, possiamo avere lo stesso numero di bucket anche .