D-Net

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

import java.io.IOException;

import javax.xml.transform.TransformerConfigurationException;

import javax.xml.transform.TransformerFactoryConfigurationError;

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;

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;

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

		} catch (final DocumentException e) {

			System.err.println("unable to parse contextMap: " + contextMap);

		}

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

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

		relClasses = new RelClasses(relClassJson);

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

	}

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

			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 TransformerConfigurationException e) {

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

			throw new RuntimeException(e);

		} catch (final TransformerFactoryConfigurationError 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) {

		try {

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

				}

			}

		} 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;

		}

	}

}