Zwei Mapper mit mehreren Datenquellen

Question

Ich bin schrieb 2 Mapper Map1 und Map2

Map1- liest eine Seq-Datei in HDFS und verarbeitet sie.

Map2 - liest von HBASE und erzeugt denselben Schlüssel, Wertpaar wie Map1.

Schließlich füge ich sie in ReducerAll zusammen.

Das Problem ist, dass nur einer der Mapper ausgeführt wird und der Job ohne Fehler abgeschlossen wird. Nur der letzte Mapper läuft (d. H. TableMapReduceUtil). Wenn ich die Zeilen TableMapReduceUtil und MultipleInputs austausche, dann läuft der letzte, d.h. MultipleInputs Mapper.

Was mache ich hier falsch? Für beide Szenarien werden keine Fehler ausgegeben. Ich lese auch 2 Dateien mit addCacheFile() für die Verarbeitung, aber das ist hier egal, denke ich.

Job job3 = Job.getInstance(config, "Test"); 
if (true) { 


    job3.setJarByClass(Main.class); 


    job3.setMapOutputKeyClass(ImmutableBytesWritable.class); 
    job3.setMapOutputValueClass(ImmutableBytesWritable.class); 
    job3.setOutputKeyClass(ImmutableBytesWritable.class); 
    job3.setOutputValueClass(ImmutableBytesWritable.class); 


    job3.getConfiguration().set("StartDate", c_startDate); 
    job3.getConfiguration().set("EndDate", c_endDate); 


    job3.addCacheFile(new URI(args[8])); 
    job3.getConfiguration().set("abc", args[8].substring(args[8].lastIndexOf("/") + 1)); 

    job3.addCacheFile(new URI(args[9])); 
    job3.getConfiguration().set("xyz", args[9].substring(args[9].lastIndexOf("/") + 1)); 
    job3.setReducerClass(ReducerAll.class); 
    job3.setOutputFormatClass(SequenceFileOutputFormat.class); 

    job3.setNumReduceTasks(10); 

    Scan scan = new Scan(); 
    scan.setAttribute(Scan.SCAN_ATTRIBUTES_TABLE_NAME, Bytes.toBytes("hbasetable")); 
    scan.setCaching(300); 
    scan.setCacheBlocks(false); 

    MultipleInputs.addInputPath(job3, new Path(args[6]), SequenceFileInputFormat.class, Map1.class); 
    TableMapReduceUtil.initTableMapperJob(
      "hbasetable", 
      scan, 
      Map2.class, 
      ImmutableBytesWritable.class, 
      ImmutableBytesWritable.class, 
      job3); 


    FileOutputFormat.setOutputPath(job3, new Path(args[7])); 
    job3.waitForCompletion(true); 
    if (!job3.waitForCompletion(true)) { 
    return (1); 
    } 

Answer 1

Ich glaube, das Verhalten aufgrund der folgenden zwei Zeilen ist: -

MultipleInputs.addInputPath(job3, new Path(args[6]), SequenceFileInputFormat.class, Map1.class); 
TableMapReduceUtil.initTableMapperJob(
     "hbasetable", 
     scan, 
     Map2.class, 
     ImmutableBytesWritable.class, 
     ImmutableBytesWritable.class, 
     job3); 

1. Es gibt nur einen Job job3
2. Obwohl Sie erwähnt haben, dass es zwei Mapper sind, Schau dir die Mappertypen an. Der Mapper in Map1 wäre anders von Map2. Map1 ist ein Mapper und Map2 ist ein TableMapper
3. halten die beiden Aussagen zusammen, bedeutet nicht, dass sie im Wesentlichen in einer MultipleInputs Setup für job3 geprügelt werden. Die MultipleInputs haben nur noch ein Setup für Map1. Die andere Einrichtung von Map2 ist noch getrennt.
4. Jetzt für die Ausführung. Letztere der beiden Konfigurationen MultipleInputs oder TableMapReduceUtil überschreibt die vorherige in der job3 und somit nur eine einzige Mapper führt

PS: - Bitte lassen Sie mich wissen, ob dies falsch ist, ich bin noch mein Verständnis zu validieren Ich habe hier auf meiner Maschine vorgestellt.