D-Net

package eu.dnetlib.data.mapreduce.hbase.index;

import java.io.IOException;

import java.util.Set;

import com.google.common.base.Splitter;

import com.google.common.collect.Sets;

import eu.dnetlib.data.mapreduce.JobParams;

import eu.dnetlib.data.mapreduce.hbase.index.config.ContextMapper;

import eu.dnetlib.data.mapreduce.hbase.index.config.EntityConfigTable;

import eu.dnetlib.data.mapreduce.hbase.index.config.IndexConfig;

import eu.dnetlib.data.mapreduce.hbase.index.config.RelClasses;

import eu.dnetlib.data.mapreduce.util.OafDecoder;

import eu.dnetlib.data.mapreduce.util.OafHbaseUtils;

import eu.dnetlib.data.mapreduce.util.OafRowKeyDecoder;

import eu.dnetlib.data.mapreduce.util.XmlRecordFactory;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

import org.dom4j.DocumentException;

public class PrepareFeedReducer extends Reducer<Text, ImmutableBytesWritable, Text, Text> {

	private static final Log log = LogFactory.getLog(PrepareFeedReducer.class); // NOPMD by marko on 11/24/08 5:02 PM

	private EntityConfigTable entityConfigTable;

	private ContextMapper contextMapper = new ContextMapper();

	private RelClasses relClasses;

	private Set<String> otherDatasourceTypesUForUI;

	private String schemaLocation;

	private final boolean entityDefaults = true;

	private final boolean relDefaults = false;

	private final boolean childDefaults = false;

	@Override

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

		final String json = context.getConfiguration().get(JobParams.INDEX_ENTITY_LINKS);

		log.info(JobParams.INDEX_ENTITY_LINKS + ":\n" + json);

		entityConfigTable = IndexConfig.load(json).getConfigMap();

		schemaLocation = context.getConfiguration().get("oaf.schema.location");

		log.info("schema location" + ":\n" + schemaLocation);

		final String contextMap = context.getConfiguration().get("contextmap");

		log.info("contextmap:\n" + contextMap);

		try {

			contextMapper = ContextMapper.fromXml(contextMap);

			log.info("context map:\n" + contextMapper);

		} catch (final DocumentException e) {

			throw new RuntimeException("unable to parse contextMap: " + contextMap, e);

		}

		final String relClassJson = context.getConfiguration().get("relClasses");

		log.info("relClassesJson:\n" + relClassJson);

		relClasses = new RelClasses(relClassJson);

		log.info("relClasses:\n" + relClasses);

		String otherDsTypesUI = context.getConfiguration().get("ui.other.datasourcetypes");

		log.info("ui.other.datasourcetypes:" + otherDsTypesUI);

		otherDatasourceTypesUForUI = Sets.newHashSet(Splitter.on(",").trimResults().omitEmptyStrings().split(otherDsTypesUI));

		log.info("ui.other.datasourcetypes parsed:" + otherDatasourceTypesUForUI);

	}

	@Override

	protected void reduce(final Text key, final Iterable<ImmutableBytesWritable> values, final Context context) throws IOException, InterruptedException {

		final OafRowKeyDecoder keyDecoder = OafRowKeyDecoder.decode(key.toString());

		try {

			final XmlRecordFactory builder =

					new XmlRecordFactory(entityConfigTable, contextMapper, relClasses, schemaLocation, entityDefaults, relDefaults, childDefaults, otherDatasourceTypesUForUI);

			try {

				decodeValues(context, keyDecoder, values, builder);

				if (builder.isValid()) {

					context.write(keyDecoder.getKeyAsText(), new Text(builder.build()));

					// context.getCounter("entity", keyDecoder.getType().toString()).increment(1);

				} else {

					// System.err.println("invalid " + getType(key) + ": " + key.toString());

					context.getCounter("missing body (reduce)", keyDecoder.getType().toString()).increment(1);

				}

			} catch (final OutOfMemoryError e) {

				context.getCounter("error", e.getClass().getName()).increment(1);

				log.error(String.format("memory error building entity\nid: '%s'\ncounters: %s", keyDecoder.getKey(), builder.getRelCounters()), e);

				throw e;

			}

		} catch (final Exception e) {

			context.getCounter("error", e.getClass().getName()).increment(1);

			throw new RuntimeException(e);

		}

	}

	private void decodeValues(final Context context,

			final OafRowKeyDecoder keyDecoder,

			final Iterable<ImmutableBytesWritable> values,

			final XmlRecordFactory builder) {

		for (final ImmutableBytesWritable bytes : values) {

			final OafDecoder decoder = OafHbaseUtils.decode(bytes);

			switch (decoder.getKind()) {

			case entity:

				builder.setMainEntity(decoder);

				break;

			case relation:

				if (decoder.getOafRel().getChild()) {

					builder.addChild(keyDecoder.getType(), decoder);

				} else {

					builder.addRelation(keyDecoder.getType(), decoder);

				}

				break;

			default:

				throw new IllegalArgumentException("unknow type: " + decoder.getKind());

			}

		}

	}

}