Comment suggéreriez-vous d'effectuer "Join" avec Hadoop streaming?

Question

J'ai deux fichiers, dans les formats suivants:

field1, field2, field3 
field4, field1, field5 

Un autre numéro de champ indique un sens différent.

Je veux rejoindre les deux fichiers en utilisant Hadoop en streaming basé sur le champ mutuel (field1 dans l'exemple ci-dessus) de sorte que la sortie sera field1, field2, field3, field4, field5 (d'autres sont ok ordonnancements tout aussi longtemps qu'ils ont tous les champs).

Answer 1

Hadoop a une bibliothèque appelée KeyFieldBasedPartitioner http://hadoop.apache.org/mapreduce/docs/r0.21.0/api/org/apache/hadoop/mapred/lib/KeyFieldBasedPartitioner.html

En utilisant cela comme une option dans le lancement de votre travail en tant que partitionneur pour votre travail en continu permet de briser votre sortie mappeur en paires clé/valeur et ont les clés se hachés en place ensemble va même réducteur et de tri, y compris les valeurs http://hadoop.apache.org/mapreduce/docs/r0.21.0/streaming.html#More+Usage+Examples

$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar \ 
-D stream.map.output.field.separator=. \ 
-D stream.num.map.output.key.fields=4 \ 
-D mapreduce.map.output.key.field.separator=. \ 
-D mapreduce.partition.keypartitioner.options=-k1,2 \ 
-D mapreduce.job.reduces=12 \ 
-input myInputDirs \ 
-output myOutputDir \ 
-mapper org.apache.hadoop.mapred.lib.IdentityMapper \ 
-reducer org.apache.hadoop.mapred.lib.IdentityReducer \ 
-partitioner org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner 

ici, -D stream.map.output.field.separator =. et -D stream.num.map.output.key.fields = 4 sont expliqués ici http://hadoop.apache.org/mapreduce/docs/r0.21.0/streaming.html#Customizing+How+Lines+are+Split+into+Key%2FValue+Pairs, ils sont essentiellement comment vous avez sorti vos champs de mappeur pour définir les paires clé/valeur.

Les clés de sortie de carte du travail MapReduce ci-dessus ont normalement quatre champs séparés par ".". Toutefois, l'infrastructure MapReduce partitionnera les sorties de la carte par les deux premiers champs des clés à l'aide de l'option -D mapreduce.partition.keypartitioner.options = -k1,2. Ici, -D mapreduce.map.output.key.field.separator =. spécifie le séparateur pour la partition. Ceci garantit que toutes les paires clé/valeur ayant les mêmes deux premiers champs dans les clés seront partitionnées dans le même réducteur. Ceci est effectivement équivalent à spécifier les deux premiers champs comme clé primaire et les deux champs suivants comme secondaire. La clé primaire est utilisée pour le partitionnement et la combinaison des clés principale et secondaire est utilisée pour le tri. Pour faire une jointure, il est aussi simple que de sortir les champs de votre mappeur et de définir les options sur votre lancement de configuration pour les champs qui sont les clés et le réducteur aura toutes vos valeurs jointes par clé de manière appropriée . Si vous voulez prendre des données de plusieurs sources, continuez à ajouter plus d'entrées sur la ligne de commande ... si elles sont différentes, alors vous pouvez le reconnaître dans votre mappeur et créer une sortie au format standard à partir du mappeur.

Answer 2

Cascading fournit des abstractions utiles pour le filtrage, la jonction et le regroupement par champ. https://stackoverflow.com/questions/4626356 liens vers un exemple utile d'utilisation du streaming Hadoop en cascade.