D-Net

package eu.dnetlib.data.mapreduce.hbase.lodExport.linkage;

import org.aksw.limes.core.execution.engine.ExecutionEngine;

import org.aksw.limes.core.execution.engine.ExecutionEngineFactory;

import org.aksw.limes.core.execution.planning.plan.NestedPlan;

import org.aksw.limes.core.execution.planning.planner.ExecutionPlannerFactory;

import org.aksw.limes.core.execution.planning.planner.IPlanner;

import org.aksw.limes.core.execution.rewriter.Rewriter;

import org.aksw.limes.core.execution.rewriter.RewriterFactory;

import org.aksw.limes.core.io.cache.MemoryCache;

import org.aksw.limes.core.io.config.KBInfo;

import org.aksw.limes.core.io.config.reader.xml.XMLConfigurationReader;

import org.aksw.limes.core.io.ls.LinkSpecification;

import org.aksw.limes.core.io.mapping.Mapping;

import org.aksw.limes.core.io.preprocessing.Preprocessor;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

import org.apache.log4j.Logger;

import redis.clients.jedis.Jedis;

import redis.clients.jedis.JedisPool;

import redis.clients.jedis.exceptions.JedisException;

import java.io.BufferedReader;

import java.io.IOException;

import java.io.InputStreamReader;

import java.util.Iterator;

public class LimesReducer extends Reducer<Text, Text, Text, Text> {

	private Logger log = Logger.getLogger(LimesReducer.class);

	//address of your redis server

	private String redisHost;

	private Integer redisPort;

	private org.aksw.limes.core.io.config.Configuration config ;

	//the jedis connection pool..

	private static JedisPool pool = null;

	public static enum ENTITIES_COUNTER {

		TARGET_ENTITIES,

		SOURCE_ENTITIES,

		WRITTEN_OUT_ENTITIES

	}

	@Override

	protected void setup(Context context) throws IOException, InterruptedException {

		redisHost = context.getConfiguration().get("lod.redisHost");

		redisPort = Integer.parseInt(context.getConfiguration().get("lod.redisPort"));

		pool = new JedisPool(redisHost, redisPort);

		String configXML = context.getConfiguration().get("lod.configXML");

		String limesDTD = context.getConfiguration().get("lod.limesDTD");

		XMLConfigurationReader reader = new XMLConfigurationReader();

		config = reader.read(configXML, limesDTD);

		System.out.println("Lod config" + config.toString());

	}

	@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();

		KBInfo sourceKb = config.getSourceInfo();

		KBInfo targetKb = config.getTargetInfo();

		System.out.println("sourceKb " + config.getSourceInfo());

		System.out.println("sourceKb  func" + config.getSourceInfo().getFunctions());

		System.out.println("targetKb  func" + config.getTargetInfo());

		System.out.println("sourceKb  func" + config.getTargetInfo().getFunctions());

		Iterator<Text> it = values.iterator();

		String[] recordIds = key.toString().split(",");

		//System.out.println("recordIds" + key);

		try {

			//each list is a block

			//while (it.hasNext())

			for (int i = 1; i < recordIds.length; i++)

			{

				String recordId = recordIds[i];

				//System.out.println("recordId " +recordId);

					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])){

						if (recordId.startsWith("source_")) {

							for(String propertyDub : sourceKb.getFunctions().get(split[0]).keySet()) {

								String processedValue = Preprocessor.process(split[1], sourceKb.getFunctions().get(split[0]).get(propertyDub));

								sourceCache.addTriple(subject, propertyDub, processedValue);

							}

							//System.out.println(subject + "," + split[0] + "," + split[1]);

							//log.info(subject + "," + split[0] + "," + split[1]);

							context.getCounter(ENTITIES_COUNTER.SOURCE_ENTITIES).increment(1);

						} else {

							for(String propertyDub : targetKb.getFunctions().get(split[0]).keySet()) {

									String processedValue = Preprocessor.process(split[1], targetKb.getFunctions().get(split[0]).get(propertyDub));

									targetCache.addTriple(subject, propertyDub, processedValue);

								}

							context.getCounter(ENTITIES_COUNTER.TARGET_ENTITIES).increment(1);

						}

					}

						//		System.out.println(subject + "," + split[0] + "," + split[1]);

						//context.write(new Text(subject), new Text(split[0] + "," + split[1]));

					}

			}

			//link specification

			Rewriter rw = RewriterFactory.getRewriter("Default");

			LinkSpecification ls = new LinkSpecification(config.getMetricExpression(), config.getVerificationThreshold());

			LinkSpecification rwLs = rw.rewrite(ls);

			//planning

			IPlanner planner = ExecutionPlannerFactory.getPlanner(config.getExecutionPlan(), sourceCache, targetCache);

			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)

			//output

			for (String source : acceptanceMapping.getMap().keySet()) {//gia kathe source blepoume ta targets

				for (String target : acceptanceMapping.getMap().get(source).keySet()) {//gia kathe target blepoume to confidence

					//System.out.println(" ouput "+ source + " " + acceptanceMapping.getConfidence(source, target) + " " + target);

					context.write(new Text(source), new Text(target + "," + acceptanceMapping.getConfidence(source, target)));

					//mporeis na grapseis source-target se osous syndiasmous vgei! Edw sou deixnw pws grafeis kai confidence mesa gia na kseroume

					//ayta apo olous tous reducers kalo 8a itan na grafonte se ena arxeio..LinkSet 8a to onomazoume kai 8a exei ola ta links pou 8a vgazei to limes

					context.getCounter(ENTITIES_COUNTER.WRITTEN_OUT_ENTITIES).increment(1);

				}

			}

		} catch (Exception e) {

			log.error(e.toString(), e);

			//throw new InterruptedException(e.toString());

		}

	}

	private String getFromRedis(String key) throws Exception {

		//get a jedis connection jedis connection pool

		Jedis jedis = pool.getResource();

		try {

			Iterator iterator = jedis.smembers(key).iterator();

			if (iterator.hasNext()) {

				String fields = (String) iterator.next();

				//System.out.println("Redis fields " + fields);

				return fields;

			}

			return "";

		} catch (JedisException e) {

			log.error("Error writing entity to Redis", e);

			if (null != jedis) {

				pool.returnBrokenResource(jedis);

				jedis = null;

			}

			throw new RuntimeException("Error writing to Redis  ", e);

		} finally

		{

			if (null != jedis) pool.returnResource(jedis);

		}

	}

	private String readFile(String path) throws IOException {

		FileSystem fs = FileSystem.get(new Configuration());

		Path pt = new Path(path);

		BufferedReader br = new BufferedReader(new InputStreamReader(fs.open(pt)));

		String line = br.readLine();

		while (line != null) {

			line += br.readLine();

		}

		br.close();

		return line;

	}

	@Override

	protected void cleanup(Context context) throws IOException, InterruptedException {

		log.info("Cleaning up reducer...");

	}

	public String getRedisHost() {

		return redisHost;

	}

	public void setRedisHost(String redisHost) {

		this.redisHost = redisHost;

	}

	public Integer getRedisPort() {

		return redisPort;

	}

	public void setRedisPort(Integer redisPort) {

		this.redisPort = redisPort;

	}

	public static JedisPool getPool() {

		return pool;

	}

	public static void setPool(JedisPool pool) {

		LimesReducer.pool = pool;

	}

}