D-Net

package eu.dnetlib.data.mapreduce.hbase.dedup.cc;

import java.io.IOException;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

/**

 * Created by claudio on 14/10/15.

 */

public class MindistSearchMapper extends Mapper<Text, VertexWritable, Text, VertexWritable> {

	private static final Log log = LogFactory.getLog(MindistSearchMapper.class);

	private boolean debug = false;

	@Override

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

		super.setup(context);

		debug = context.getConfiguration().getBoolean("mindist_DEBUG", false);

		log.info("debug mode: " + debug);

	}

	@Override

	protected void map(Text key, VertexWritable value, Context context) throws IOException, InterruptedException {

		emit(key, value, context);

		if (value.isActivated()) {

			VertexWritable vertex = new VertexWritable();

			for (Text edge : value.getEdges()) {

				if (!edge.toString().equals(value.getVertexId().toString())) {

					vertex.setVertexId(value.getVertexId());

					vertex.setEdges(null);

					emit(edge, vertex, context);

				}

			}

		}

	}

	private void emit(final Text key, final VertexWritable vertex, final Context context) throws IOException, InterruptedException {

		context.write(key, vertex);

		if (debug) {

			log.info(vertex.toJSON());

		}

	}

}