package es.tid.ps.dynamicprofile.categoryextraction;

import java.io.IOException;

import org.apache.hadoop.mapreduce.Reducer;

import es.tid.ps.dynamicprofile.dictionary.DictionaryHandler;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.io.NullWritable;

public class CategoryExtractionReducer extends
        Reducer<CompositeKey, NullWritable, CompositeKey, CategoryInformation> {
    private final static String CATEGORY_DELIMITER = "/";

    @Override
    public void setup(Context context) throws IOException {
        DictionaryHandler.init(DistributedCache.getLocalCacheFiles(
                context.getConfiguration()));
    }

    @Override
    protected void reduce(CompositeKey key, Iterable<NullWritable> values,
            Context context) throws IOException, InterruptedException {
        // Use the comScore API
        String url = key.getSecondaryKey();
        String categories = DictionaryHandler.getUrlCategories(url);
        if (categories.isEmpty()) {
            context.getCounter(CategoryExtractionCounter.EMPTY_CATEGORY).increment(1L);
        } else {
            context.getCounter(CategoryExtractionCounter.VALID_CATEGORY).increment(1L);
        }

        // Count the number of instances of the same URL
        long count = 0;
        while (values.iterator().hasNext()) {
            count++;
        }

        CategoryInformation cat = new CategoryInformation(key.getPrimaryKey(),
                key.getSecondaryKey(), count,
                categories.split(CATEGORY_DELIMITER));
        context.write(key, cat);
    }
}

@@ -989,16 +989,38 @@
 tegories
+ == null %7C%7C categories
 .isEmpty
@@ -1021,24 +1021,24 @@
 sEmpty()) %7B%0A
-
             
@@ -1125,21 +1125,31 @@
     
-%7D else %7B%0A    
+    return;%0A        %7D%0A%0A
     
@@ -1227,26 +1227,16 @@
 ent(1L);
-%0A        %7D
 %0A%0A      

stop for null categories