package com.ontology2.telepath.mergeAgainstDBpedia;

import com.ontology2.bakemono.joins.TaggedTextItem;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class MergeReducer extends Reducer<TaggedTextItem,TaggedTextItem,Text,Text> {

    @Override
    protected void reduce(TaggedTextItem key, Iterable<TaggedTextItem> values, Context context) throws IOException, InterruptedException {
        Text toValue=null;
        for(TaggedTextItem value:values) {
            int tag=value.getTag().get();
            switch(tag) {
                case 1:
                    toValue=key.getKey();      // we can write the key because it exists
                    break;
                case 2:
                    toValue=new Text(value.getKey());    // we will write the value because we redirect to it
                    break;
                case 3:
                    if(toValue!=null)
                    context.write(toValue,value.getKey());
            }
        }
    }
}

@@ -46,16 +46,56 @@
 pedia;%0A%0A
+import com.google.common.base.Splitter;%0A
 import c
@@ -138,16 +138,101 @@
 xtItem;%0A
+import org.apache.commons.logging.Log;%0Aimport org.apache.commons.logging.LogFactory;%0A
 import o
@@ -415,16 +415,68 @@
 ,Text%3E %7B
+%0A    Log LOG= LogFactory.getLog(MergeReducer.class);
 %0A%0A    @O
@@ -653,50 +653,237 @@
 ll;%0A
-        for(TaggedTextItem value:values) %7B
+%0A        LOG.info(%22Processed key %5B%22+key.getKey()+%22%5D with tag %5B%22+key.getTag().get()+%22%5D%22);%0A        for(TaggedTextItem value:values) %7B%0A            LOG.info(%22Processed value %5B%22+value.getKey()+%22%5D with tag %5B%22+value.getTag().get()+%22%5D%22);
 %0A   

add INFO level logging to MergeReducer so we can see all data going through. (will run this against a tiny data set)