D-Net

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

import java.io.IOException;

import java.util.List;

import com.google.common.base.Function;

import com.google.common.collect.Iterables;

import com.google.common.collect.Lists;

import eu.dnetlib.data.mapreduce.JobParams;

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

import eu.dnetlib.data.proto.OafProtos.Oaf;

import eu.dnetlib.data.proto.PersonProtos.Person;

import eu.dnetlib.data.proto.PersonProtos.Person.CoAuthor;

import eu.dnetlib.data.proto.TypeProtos.Type;

import eu.dnetlib.data.transform.xml.AbstractDNetXsltFunctions;

import eu.dnetlib.miscutils.maps.ConcurrentSizedMap;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.hadoop.hbase.client.Put;

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

import org.apache.hadoop.hbase.mapreduce.TableReducer;

import org.apache.hadoop.hbase.util.Bytes;

public class CoAuthorReducer extends TableReducer<ImmutableBytesWritable, ImmutableBytesWritable, ImmutableBytesWritable> {

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

	private static int MAX_COAUTHORS = 50;

	private int max_coauthors = MAX_COAUTHORS;

	@Override

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

		max_coauthors = Integer.parseInt(context.getConfiguration().get("max.coauthors"));

	}

	@Override

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

			InterruptedException {

		final byte[] keyBytes = key.copyBytes();

		final String rowKey = new String(keyBytes);

		log.info("Reducing key: '" + rowKey + "'");

		final Function<Person, CoAuthor> personToCoAuthor = new Function<Person, CoAuthor>() {

			@Override

			public CoAuthor apply(final Person p) {

				final CoAuthor.Builder cb = CoAuthor.newBuilder();

				cb.setId(getId(rowKey, p.getMetadata().getFullname().getValue().trim()));

				cb.getMetadataBuilder().mergeFrom(p.getMetadata());

				return cb.build();

			}

		};

		final Oaf.Builder out = Oaf.newBuilder();

		final ConcurrentSizedMap<String, CoAuthor> coAuthors = new ConcurrentSizedMap<String, CoAuthor>();

		coAuthors.setQueueSize(max_coauthors);

		for (final ImmutableBytesWritable ibw : values) {

			final OafDecoder d = OafDecoder.decode(ibw.copyBytes());

			final Type type = d.getEntity().getType();

			switch (type) {

			case person:

				out.mergeFrom(d.getOaf());

				out.getEntityBuilder().getPersonBuilder().clearCoauthor();

				context.getCounter("coauthor", "merge").increment(1);

				break;

			case result:

				final List<Person> authorList = d.getEntity().getResult().getAuthorList();

				for (final CoAuthor coauthor : Lists.newArrayList(Iterables.transform(authorList, personToCoAuthor))) {

					coAuthors.put(coauthor.getId(), incrementCount(coAuthors, coauthor));

				}

				break;

			case datasource:

			case organization:

			case project:

			default:

				context.getCounter("coauthor", "skipped entity type (reduce)").increment(1);

				break;

			}

		}

		coAuthors.remove(rowKey); // remove himself

		out.getEntityBuilder().getPersonBuilder().addAllCoauthor(coAuthors.values());

		final Put put = new Put(keyBytes).add(Bytes.toBytes(Type.person.toString()), Bytes.toBytes("body"), out.build().toByteArray());

		put.setWriteToWAL(JobParams.WRITE_TO_WAL);

		context.write(key, put);

	}

	private CoAuthor incrementCount(final ConcurrentSizedMap<String, CoAuthor> coAuthors, final CoAuthor coauthor) {

		if (coAuthors.containsKey(coauthor.getId())) {

			final CoAuthor.Builder c = CoAuthor.newBuilder(coAuthors.get(coauthor.getId()));

			c.setCoauthoredpapers(c.getCoauthoredpapers() + 1);

			return c.build();

		}

		return coauthor;

	}

	private String getId(final String resultId, final String name) {

		final String prefix = StringUtils.substringBefore(StringUtils.substringAfter(resultId, "|"), "::");

		// final String id = StringUtils.substringAfter(resultId, "|") + "::" + name.replaceAll("\\s+", " ").trim();

		final String id = name.replaceAll("\\s+", " ").trim();

		return AbstractDNetXsltFunctions.oafId(Type.person.toString(), prefix, id);

	}

}