Index: modules/dnet-openaire-lodinterlinking/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/build/BlockReducer.java
===================================================================
--- modules/dnet-openaire-lodinterlinking/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/build/BlockReducer.java	(revision 43484)
+++ modules/dnet-openaire-lodinterlinking/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/build/BlockReducer.java	(revision 43485)
@@ -11,9 +11,8 @@
 import org.apache.log4j.Logger;
 
 public class BlockReducer extends Reducer<Text, Text, Text, Text> {
-    public static enum ENTITIES_COUNTER {
-
-        TOTAL_ENTITIES,
+    public static enum BLOCKS_COUNTER {
+        TOTAL_ENTITIES
     }
 
     private Logger log = Logger.getLogger(BlockReducer.class);
@@ -40,16 +39,12 @@
                 nunberOfEntities++;
             }
 
+            MultipleOutputWriter.write("b", key, new Text(field.toString().getBytes(Charset.forName("UTF-8"))), "blocks/b");
 
-            MultipleOutputWriter.write("b", key,
-                    new Text(field.toString().getBytes(Charset.forName("UTF-8"))), "blocks/b");
+            MultipleOutputWriter.write("entitiesNumber", key, new Text(String.valueOf(nunberOfEntities).getBytes(Charset.forName("UTF-8"))), "stats/entitiesNumber");
 
-            //  context.write(key, new Text(field.toString()));
-            MultipleOutputWriter.write("entitiesNumber", key,
-                    new Text(String.valueOf(nunberOfEntities).getBytes(Charset.forName("UTF-8"))), "stats/entitiesNumber");
 
-
-            context.getCounter(ENTITIES_COUNTER.TOTAL_ENTITIES).increment(1);
+            context.getCounter(BLOCKS_COUNTER.TOTAL_ENTITIES).increment(1);
         } catch (Exception e) {
             throw new InterruptedException(e.getMessage());
         }
Index: modules/dnet-openaire-lodinterlinking/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/build/SourceBuildMapper.java
===================================================================
--- modules/dnet-openaire-lodinterlinking/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/build/SourceBuildMapper.java	(revision 43484)
+++ modules/dnet-openaire-lodinterlinking/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/build/SourceBuildMapper.java	(revision 43485)
@@ -40,7 +40,7 @@
     private LodConfiguration lodConfiguration;
     private String lastExecutionDate;
 
-    public static enum ENTITIES_COUNTER {
+    public static enum SOURCE_BUILD_COUNTERS {
 
         BLOCKING_KEYS,
         REDIS_RECORDS
@@ -85,7 +85,7 @@
             String id = "source_" + (String) ((List) Fields.get("id")).get(0);
 
             for (Map.Entry<String, String> entry : Fields.entries()) {
-                 //  if (lodConfiguration.isValidField(entityType, entry.getKey().replace("/", "\\/")))
+                //  if (lodConfiguration.isValidField(entityType, entry.getKey().replace("/", "\\/")))
                 {
                     value.append(entry.getKey() + "\t" + entry.getValue() + ",");
                 }
@@ -94,7 +94,7 @@
 
             //Write to cache
             writeToRedis(id, value.toString());
-            context.getCounter(ENTITIES_COUNTER.REDIS_RECORDS).increment(1);
+            context.getCounter(SOURCE_BUILD_COUNTERS.REDIS_RECORDS).increment(1);
 
             ArrayList<String> blockingKeys = Blocking.tokenBlocking(Fields);
 
@@ -101,7 +101,7 @@
             for (String blockingKey : blockingKeys) {
                 //Write BlockingKey, RecordID to output
                 context.write(new Text(blockingKey), new Text(id));
-                context.getCounter(ENTITIES_COUNTER.BLOCKING_KEYS).increment(1);
+                context.getCounter(SOURCE_BUILD_COUNTERS.BLOCKING_KEYS).increment(1);
             }
 
         } catch (Exception e) {
@@ -116,18 +116,11 @@
     private void writeToRedis(String key, String value) throws Exception {
 
         try {
-
-
             connection.set(key, value);
-            //   log.info("Set to redis " + " " + key + "to " + value);
-
         } catch (Exception e) {
-
             log.error("Error writing entity to Redis", e);
-
             throw new RuntimeException("Error writing to Redis  ", e);
         }
-
     }
 
 
@@ -137,7 +130,6 @@
             connection.close();
         }
         super.cleanup(context);
-
     }
 
 
Index: modules/dnet-openaire-lodinterlinking/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/build/TargetBuildMapper.java
===================================================================
--- modules/dnet-openaire-lodinterlinking/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/build/TargetBuildMapper.java	(revision 43484)
+++ modules/dnet-openaire-lodinterlinking/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/build/TargetBuildMapper.java	(revision 43485)
@@ -42,7 +42,7 @@
     private RedisClient client;
     private RedisConnection connection;
 
-    public static enum ENTITIES_COUNTER {
+    public static enum TARGET_BUILD_COUNTERS {
 
         BLOCKING_KEYS,
         REDIS_RECORDS
@@ -55,7 +55,6 @@
         lodConfiguration.load(context.getConfiguration().get("lod.targetMappings"));
         redisHost = context.getConfiguration().get("lod.redisHost");
         redisPort = Integer.parseInt(context.getConfiguration().get("lod.redisPort"));
-
         initLettuceRedis();
     }
 
@@ -62,7 +61,6 @@
     private void initLettuceRedis() {
         client = RedisClient.create("redis://" + redisHost + ":" + redisPort);
         connection = client.connect();
-        System.out.println("Connected to Redis" + "redis://" + redisHost + ":" + redisPort);
         log.info("Connected to Redis" + "redis://" + redisHost + ":" + redisPort);
     }
 
@@ -98,15 +96,14 @@
             }
             //Write to cache
             writeToRedis(id, value.toString());
-            context.getCounter(ENTITIES_COUNTER.REDIS_RECORDS).increment(1);
+            context.getCounter(TARGET_BUILD_COUNTERS.REDIS_RECORDS).increment(1);
 
             ArrayList<String> blockingKeys = Blocking.tokenBlocking(Fields);
 
             for (String blockingKey : blockingKeys) {
                 //Write BlockingKey, RecordID to output
-                // log.info("Writing out " + blockingKey + " , " + id);
                 context.write(new Text(blockingKey), new Text(id));
-                context.getCounter(ENTITIES_COUNTER.BLOCKING_KEYS).increment(1);
+                context.getCounter(TARGET_BUILD_COUNTERS.BLOCKING_KEYS).increment(1);
             }
 
         } catch (Exception e) {
@@ -114,13 +111,11 @@
             log.error("result error    " + result.toString());
             throw new RuntimeException(e);
         }
-
     }
 
     private void writeToRedis(String key, String value) throws Exception {
 
         try {
-        /*    13611*/
             connection.set(key, value);
         } catch (Exception e) {
             log.error("Error writing entity to Redis", e);
Index: modules/dnet-openaire-lodinterlinking/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/utils/DatasetComparator.java
===================================================================
--- modules/dnet-openaire-lodinterlinking/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/utils/DatasetComparator.java	(nonexistent)
+++ modules/dnet-openaire-lodinterlinking/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/utils/DatasetComparator.java	(revision 43485)
@@ -0,0 +1,116 @@
+package eu.dnetlib.data.mapreduce.hbase.lodExport.utils;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.LocatedFileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RemoteIterator;
+import org.json.simple.JSONObject;
+import org.json.simple.parser.JSONParser;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.math.BigInteger;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.TreeMap;
+
+/**
+ * Created by eri_k on 5/26/2016.
+ */
+public class DatasetComparator {
+
+    public static void main(String[] args) throws Exception {
+
+        String statsInputPath = "/tmp/lod_blocks/stats/";
+        String groundTruthPath = "/tmp/lod_zenodo/";
+        String linkageOutputPath = "/tmp/lod_final/";
+
+
+        try {
+
+            FileSystem fs = FileSystem.get(new Configuration());
+            List linkageFiles = getFiles(fs, new Path(linkageOutputPath));
+            HashMap linkageMap = getLinkageRecordMap(fs, linkageFiles);
+
+            List groundTruthFiles = getFiles(fs, new Path(groundTruthPath));
+            HashMap groundTruthMap = getGroundRecordMap(fs, groundTruthFiles);
+
+
+        } catch (Exception e) {
+
+            throw new Exception("Error : ", e);
+        }
+    }
+
+    private static List getFiles(FileSystem fs, Path path) throws IOException {
+        RemoteIterator<LocatedFileStatus> Files;
+        List<String> fileNames = new ArrayList<String>();
+
+        Files = fs.listFiles(path, false);
+
+        while (Files.hasNext()) {
+            String fileName = Files.next().getPath().toString();
+            fileNames.add(fileName);
+
+        }
+        System.out.println("Filenames" + fileNames);
+        return fileNames;
+    }
+
+
+
+
+    private static HashMap<String, String> getLinkageRecordMap(FileSystem fs, List<String> fileNames) throws Exception {
+        String line = null;
+        try {
+            HashMap<String, String> recordsMap = new HashMap<>();
+            for (String file : fileNames) {
+                BufferedReader br = new BufferedReader(new InputStreamReader(fs.open(new Path(file))));
+                line = br.readLine();
+                while (line != null) {
+                    String[] split = line.split("\t");
+                    String record1 = split[0];
+                    String record2 = split[0].substring(0, split[0].indexOf(","));
+                    recordsMap.put(record1, record2);
+                    line = br.readLine();
+                }
+            }
+            return recordsMap;
+        } catch (Exception e) {
+            throw new Exception("Error : " + line, e);
+        }
+
+    }
+
+
+//ground truth sample
+//<http://lod.openaire.eu/data/result/doajarticles::57a7815dea6ce88034f6cd9e85abf08b> <http://www.w3.org/2002/07/owl#sameAs> <http://dblp.l3s.de/d2r/resource/publications/journals/bmcbi/KestlerMKBGLKZW08>
+
+    private static HashMap<String, String> getGroundRecordMap(FileSystem fs, List<String> fileNames) throws Exception {
+        String line = null;
+        try {
+            HashMap<String, String> recordsMap = new HashMap<>();
+            for (String file : fileNames) {
+                BufferedReader br = new BufferedReader(new InputStreamReader(fs.open(new Path(file))));
+                line = br.readLine();
+                while (line != null) {
+                    String[] split = line.split(" ");
+                    String record1 = split[0];
+                    String record2 = split[2];
+                    recordsMap.put(record1, record2);
+                    line = br.readLine();
+                }
+            }
+            return recordsMap;
+        } catch (Exception e) {
+            throw new Exception("Error : " + line, e);
+        }
+
+    }
+
+}
+
+
Index: modules/dnet-openaire-lodinterlinking/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/preprocessing/DatasetReducer.java
===================================================================
--- modules/dnet-openaire-lodinterlinking/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/preprocessing/DatasetReducer.java	(revision 43484)
+++ modules/dnet-openaire-lodinterlinking/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/preprocessing/DatasetReducer.java	(revision 43485)
@@ -47,16 +47,11 @@
             String output = entityType;
             String outputFile = dataset + "/" + output;
 
-            System.out.println("output " + output);
-            System.out.println("output file " + dataset + "/" + output);
-
-            MultipleOutputWriter.write(output, new Text("id," + id),
-                    new Text(value.toString().getBytes(Charset.forName("UTF-8"))), outputFile);
+            MultipleOutputWriter.write(output, new Text("id," + id), new Text(value.toString().getBytes(Charset.forName("UTF-8"))), outputFile);
             context.getCounter(ENTITIES_COUNTER.TOTAL_ENTITIES).increment(1);
         } catch (Exception e) {
             log.error(entityType + e.toString());
             throw new InterruptedException(entityType + " " + id + " " + e.toString());
-
         }
     }
 
@@ -64,6 +59,5 @@
     protected void cleanup(Context context) throws IOException, InterruptedException {
         log.info("Cleaning up reducer...");
         MultipleOutputWriter.close();
-
     }
 }
Index: modules/dnet-openaire-lodinterlinking/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/preprocessing/SourceMapper.java
===================================================================
--- modules/dnet-openaire-lodinterlinking/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/preprocessing/SourceMapper.java	(revision 43484)
+++ modules/dnet-openaire-lodinterlinking/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/preprocessing/SourceMapper.java	(revision 43485)
@@ -31,7 +31,7 @@
 
     private LodConfiguration lodConfiguration;
 
-    public static enum ENTITIES_COUNTER {
+    public static enum SOURCE_COUNTERS {
 
         SOURCE_ENTITIES,
     }
@@ -95,7 +95,7 @@
 
                 Text key = new Text("OA" + "," + type + "," + subject);
                 context.write(key, new Text(value.toString()));
-                context.getCounter(ENTITIES_COUNTER.SOURCE_ENTITIES).increment(1);
+                context.getCounter(SOURCE_COUNTERS.SOURCE_ENTITIES).increment(1);
             }
         } catch (Exception e) {
             log.error("Error writing entity to M/R output", e);
Index: modules/dnet-openaire-lodinterlinking/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/preprocessing/TargetMapper.java
===================================================================
--- modules/dnet-openaire-lodinterlinking/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/preprocessing/TargetMapper.java	(revision 43484)
+++ modules/dnet-openaire-lodinterlinking/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/preprocessing/TargetMapper.java	(revision 43485)
@@ -20,7 +20,7 @@
 
     private LodConfiguration lodConfiguration;
 
-    public static enum ENTITIES_COUNTER {
+    public static enum TARGET_COUNTERS {
 
         TARGET_ENTITIES,
     }
@@ -82,7 +82,7 @@
 
                 Text key = new Text("TARGET" + "," + type + "," + subject);
                 context.write(key, new Text(value.toString()));
-                context.getCounter(TargetMapper.ENTITIES_COUNTER.TARGET_ENTITIES).increment(1);
+                context.getCounter(TargetMapper.TARGET_COUNTERS.TARGET_ENTITIES).increment(1);
             }
         } catch (Exception e) {
             log.error("Error writing entity to M/R output", e);
Index: modules/dnet-openaire-lodinterlinking/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/linkage/LinkageMapper.java
===================================================================
--- modules/dnet-openaire-lodinterlinking/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/linkage/LinkageMapper.java	(revision 43484)
+++ modules/dnet-openaire-lodinterlinking/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/linkage/LinkageMapper.java	(revision 43485)
@@ -21,7 +21,7 @@
 
     private final String OPTIMAL_BLOCKS = "optimalBlockSize";
 
-    public static enum ENTITIES_COUNTER {
+    public static enum LINKAGE_COUNTERS {
         DISCARDED_BLOCKS,
         TOTAL_BLOCKS,
         PASSED_BLOCKS
@@ -50,23 +50,23 @@
             /// /input is blocking key, string of record Ids
             //for now just output to reduce
 
-            System.out.println("key " + keyIn.toString() + " value " + result.toString());
-            log.info("key " + keyIn.toString() + "value " + result.toString());
+            System.out.println("key is " + keyIn.toString());
+            System.out.println("result is " + result.toString());
 
-            context.getCounter(ENTITIES_COUNTER.TOTAL_BLOCKS).increment(1);
+            context.getCounter(LINKAGE_COUNTERS.TOTAL_BLOCKS).increment(1);
 
             int optimalBlockSize = Integer.parseInt(getFromRedis(OPTIMAL_BLOCKS));
 
             System.out.println("Optimal size" + optimalBlockSize);
+
             String records[] = result.toString().split(",");
 
             if (records.length > optimalBlockSize) {
-                context.getCounter(ENTITIES_COUNTER.DISCARDED_BLOCKS).increment(1);
+                context.getCounter(LINKAGE_COUNTERS.DISCARDED_BLOCKS).increment(1);
             } else {
                 context.write(keyIn, result);
-                context.getCounter(ENTITIES_COUNTER.PASSED_BLOCKS).increment(1);
+                context.getCounter(LINKAGE_COUNTERS.PASSED_BLOCKS).increment(1);
             }
-
         } catch (Exception e) {
             log.error("Error writing entity to M/R output", e);
             throw new RuntimeException(e);
@@ -76,15 +76,11 @@
 
 
     private String getFromRedis(String key) throws Exception {
-
         try {
             return (String) connection.get(key);
         } catch (Exception e) {
-
             log.error("Error writing entity to Redis", e);
-
             throw new RuntimeException("Error writing to Redis  ", e);
-
         }
     }
 
@@ -92,24 +88,7 @@
     protected void cleanup(Context context) throws IOException, InterruptedException {
         connection.close();
         super.cleanup(context);
-    }
+}
 
-    public Logger getLog() {
-        return log;
-    }
 
-    public void setLog(Logger log) {
-        this.log = log;
-    }
-
-
-    public LodConfiguration getLodConfiguration() {
-        return lodConfiguration;
-    }
-
-    public void setLodConfiguration(LodConfiguration lodConfiguration) {
-        this.lodConfiguration = lodConfiguration;
-    }
-
-
 }
Index: modules/dnet-openaire-lodinterlinking/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/linkage/LimesReducer.java
===================================================================
--- modules/dnet-openaire-lodinterlinking/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/linkage/LimesReducer.java	(revision 43484)
+++ modules/dnet-openaire-lodinterlinking/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/linkage/LimesReducer.java	(revision 43485)
@@ -1,6 +1,7 @@
 package eu.dnetlib.data.mapreduce.hbase.lodExport.linkage;
 
 import com.lambdaworks.redis.RedisClient;
+import com.lambdaworks.redis.RedisConnection;
 import com.lambdaworks.redis.RedisStringsConnection;
 import org.aksw.limes.core.execution.engine.ExecutionEngine;
 import org.aksw.limes.core.execution.engine.ExecutionEngineFactory;
@@ -17,9 +18,11 @@
 import org.aksw.limes.core.io.preprocessing.Preprocessor;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapreduce.Reducer;
+import org.apache.hadoop.security.SaslOutputStream;
 import org.apache.log4j.Logger;
 
 import java.io.IOException;
+import java.util.Arrays;
 import java.util.Iterator;
 
 public class LimesReducer extends Reducer<Text, Text, Text, Text> {
@@ -28,13 +31,15 @@
 
     private String redisHost;
     private Integer redisPort;
+    private RedisClient client;
+    private RedisConnection connection;
 
     private org.aksw.limes.core.io.config.Configuration config;
 
 
-    public static enum ENTITIES_COUNTER {
-        TARGET_ENTITIES,
-        SOURCE_ENTITIES,
+    public static enum LIMES_COUNTERS {
+        TARGET_TRIPLES,
+        SOURCE_TRIPLES,
         WRITTEN_OUT_ENTITIES
     }
 
@@ -49,8 +54,13 @@
         String limesDTD = context.getConfiguration().get("lod.limesDTD");
         XMLConfigurationReader reader = new XMLConfigurationReader();
         config = reader.read(configXML, limesDTD);
+        initLettuceRedis();
+    }
 
-
+    private void initLettuceRedis() {
+        client = RedisClient.create("redis://" + redisHost + ":" + redisPort);
+        connection = client.connect();
+        log.info("Connected to Redis" + "redis://" + redisHost + ":" + redisPort);
     }
 
 
@@ -57,10 +67,6 @@
     @Override
     protected void reduce(final Text key, final Iterable<Text> values, final Context context) throws IOException, InterruptedException {
 
-
-
-/*
-
         MemoryCache sourceCache = new MemoryCache();
         MemoryCache targetCache = new MemoryCache();
 
@@ -67,46 +73,48 @@
         KBInfo sourceKb = config.getSourceInfo();
         KBInfo targetKb = config.getTargetInfo();
 
-        System.out.println("key " + key);
 
         Iterator<Text> it = values.iterator();
+
         try {
             while (it.hasNext()) {
 
-                String[] recordIds = it.next().toString().split(",");
+                String value = it.next().toString();
 
+                String[] recordIds = value.split(",");
 
                 for (String recordId : recordIds) {
 
                     String[] Fields = getFromRedis(recordId).split(",");
-
                     String subject = recordId.substring(recordId.indexOf("_") + 1);
 
-
                     for (int j = 0; j < Fields.length; j++) {
 
                         String[] split = Fields[j].split("\t");
+
                         if (sourceKb.getProperties().contains(split[0].replace("<", "").replace(">", ""))) {
-                            if (recordId.startsWith("source_")) {
+
+                            if (recordId.contains("source_")) {
+
                                 for (String propertyDub : sourceKb.getFunctions().get(split[0].replace("<", "").replace(">", "")).keySet()) {
                                     String processedValue = Preprocessor.process(split[1], sourceKb.getFunctions().get(split[0].replace("<", "").replace(">", "")).get(propertyDub));
                                     sourceCache.addTriple(subject, propertyDub, processedValue);
+                                    //System.out.println("added source triple" + subject + " " +propertyDub +" " +processedValue);
                                 }
-                                context.getCounter(ENTITIES_COUNTER.SOURCE_ENTITIES).increment(1);
+                                context.getCounter(LIMES_COUNTERS.SOURCE_TRIPLES).increment(1);
                             } else {
 
                                 for (String propertyDub : targetKb.getFunctions().get(split[0].replace("<", "").replace(">", "")).keySet()) {
                                     String processedValue = Preprocessor.process(split[1], targetKb.getFunctions().get(split[0].replace("<", "").replace(">", "")).get(propertyDub));
                                     targetCache.addTriple(subject, propertyDub, processedValue);
+                                    //System.out.println("added target triple" + subject + " " +propertyDub +" " +processedValue);
                                 }
-                                context.getCounter(ENTITIES_COUNTER.TARGET_ENTITIES).increment(1);
+                                context.getCounter(LIMES_COUNTERS.TARGET_TRIPLES).increment(1);
                             }
                         }
 
 
                     }
-
-
                 }
 
 
@@ -120,7 +128,6 @@
                 assert planner != null;
                 NestedPlan plan = planner.plan(rwLs);
                 ExecutionEngine engine = ExecutionEngineFactory.getEngine("Default", sourceCache, targetCache, config.getSourceInfo().getVar(), config.getTargetInfo().getVar());
-
                 Mapping verificationMapping = engine.execute(plan); //mappings for verification
                 Mapping acceptanceMapping = verificationMapping.getSubMap(config.getAcceptanceThreshold()); //mappings for acceptance (auta theloume)
 
@@ -129,8 +136,7 @@
                     for (String target : acceptanceMapping.getMap().get(source).keySet()) {//gia kathe target blepoume to confidence
 
                         context.write(new Text(source), new Text(target + "," + acceptanceMapping.getConfidence(source, target)));
-                        context.getCounter(ENTITIES_COUNTER.WRITTEN_OUT_ENTITIES).increment(1);
-
+                        context.getCounter(LIMES_COUNTERS.WRITTEN_OUT_ENTITIES).increment(1);
                     }
                 }
             }
@@ -139,18 +145,11 @@
             log.error(e.toString(), e);
             throw new InterruptedException(e.toString());
         }
-*/
-
     }
 
     private String getFromRedis(String key) throws Exception {
-/*
-        RedisClient client = RedisClient.create("redis://" + redisHost);
-        RedisStringsConnection<String, String> connection = client.connect();
 
-        log.info("get  from redis " + " " + key + " to " + connection.get(key));
-        return connection.get(key);*/
-        return null;
+        return (String) connection.get(key);
     }
 
 
@@ -157,8 +156,7 @@
     @Override
     protected void cleanup(Context context) throws IOException, InterruptedException {
         log.info("Cleaning up reducer...");
-
-
+        connection.close();
     }
 
     public String getRedisHost() {
