D-Net

package eu.dnetlib.data.mapreduce.hbase.broker.add;

import static eu.dnetlib.data.mapreduce.hbase.broker.mapping.EventFactory.asEvent;

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

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

import java.io.IOException;

import java.util.HashSet;

import java.util.Map;

import java.util.Set;

import org.apache.commons.collections.MapUtils;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.lang.math.RandomUtils;

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

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

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

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

import org.apache.hadoop.hbase.filter.FilterList;

import org.apache.hadoop.hbase.filter.FilterList.Operator;

import org.apache.hadoop.hbase.filter.PrefixFilter;

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

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

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

import org.apache.hadoop.io.Text;

import com.google.common.base.Function;

import com.google.common.collect.Iterables;

import com.google.common.collect.Maps;

import com.google.common.collect.Sets;

import eu.dnetlib.broker.objects.OpenAireEventPayload;

import eu.dnetlib.data.mapreduce.hbase.broker.Topic;

import eu.dnetlib.data.mapreduce.hbase.broker.mapping.OpenAireEventPayloadFactory;

import eu.dnetlib.data.mapreduce.hbase.broker.model.EventMessage;

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

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

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

/**

 * Created by claudio on 08/07/16.

 */

public class AdditionMapper extends TableMapper<Text, Text> {

	private Text outKey;

	private Text outValue;

	/**

	 * Map ProjectID -> Set of Organization info (id, name)

	 */

	private Map<String, Set<EntityInfo>> projectOrganization;

	/**

	 * Map OrganizationID -> Set of Datasource info (id, name)

	 */

	private Map<String, Set<EntityInfo>> organizationDatasource;

	private Set<String> organizationPrefixBlacklist = Sets.newHashSet();

	// White list for datasource typologies.

	private Set<String> dsTypeWhitelist = Sets.newHashSet();

	@Override

	protected void setup(final Context context) throws IOException {

		organizationPrefixBlacklist = Sets.newHashSet("nsf_________");

		dsTypeWhitelist.addAll(getPropertyValues(context, "broker.datasource.type.whitelist"));

		projectOrganization = getRelMap(context, "20", "organization", "projectOrganization_participation_isParticipant", organizationPrefixBlacklist);

		organizationDatasource = getRelMap(context, "10", "datasource", "datasourceOrganization_provision_provides", dsTypeWhitelist);

		outKey = new Text("");

		outValue = new Text();

	}

	class EntityInfo {

		private String id;

		private String name;

		public EntityInfo(final String id, final String name) {

			this.id = id;

			this.name = name;

		}

		@Override

		public int hashCode() {

			return getId().hashCode();

		}

		@Override

		public boolean equals(final Object obj) {

			return getId().equals(obj);

		}

		public String getId() {

			return id;

		}

		public void setId(final String id) {

			this.id = id;

		}

		public String getName() {

			return name;

		}

		public void setName(final String name) {

			this.name = name;

		}

	}

	@Override

	protected void map(final ImmutableBytesWritable key, final Result value, final Context context) throws IOException, InterruptedException {

		try {

			final Map<byte[], byte[]> map = value.getFamilyMap(Bytes.toBytes("result"));

			final byte[] bodyB = map.get(Bytes.toBytes("body"));

			if (MapUtils.isEmpty(map) || (bodyB == null)) {

				context.getCounter("result", "empty body").increment(1);

				return;

			}

			final Oaf oaf = Oaf.parseFrom(bodyB);

			if (oaf.getDataInfo().getDeletedbyinference()) {

				context.getCounter("result", "deletedbyinference = true").increment(1);

				return;

			}

			final Set<String> currentDatasourceIds = Sets.newHashSet(listKeys(oaf.getEntity().getCollectedfromList()));

			final Map<byte[], byte[]> resultProject = value.getFamilyMap(Bytes.toBytes("resultProject_outcome_isProducedBy"));

			if (!MapUtils.isEmpty(resultProject)) {

				for (final String projectId : asStringID(resultProject.keySet())) {

					final Set<EntityInfo> organizations = projectOrganization.get(projectId);

					if ((organizations != null) && !organizations.isEmpty()) {

						for (final EntityInfo organization : organizations) {

							final Set<EntityInfo> datasources = organizationDatasource.get(organization.getId());

							if ((datasources != null) && !datasources.isEmpty()) {

								for (final EntityInfo datasource : datasources) {

									if (!currentDatasourceIds.contains(datasource.getId())) {

										// emit event for datasourceId

										final float trust = RandomUtils.nextFloat();

										final OpenAireEventPayload payload = OpenAireEventPayloadFactory.fromOAF(oaf.getEntity(), oaf.getEntity(), trust);

										// event.setPayload(HighlightFactory.highlightEnrichPid(payload,

										// Lists.newArrayList(pids)).toJSON());

										final EventMessage event =

												asEvent(oaf.getEntity(), Topic.ADD_BY_PROJECT, payload, datasource.getId(), datasource.getName(), trust);

										emit(event, context);

										context.getCounter("event", Topic.ADD_BY_PROJECT.getValue()).increment(1);

									}

								}

							}

						}

					}

				}

			}

		} catch (final Exception e) {

			throw new RuntimeException(e);

		}

	}

	private void emit(final EventMessage e, final Context context) throws IOException, InterruptedException {

		// tKey.set(e.getMap().get("id"));

		outValue.set(e.toString());

		context.write(outKey, outValue);

	}

	private Iterable<String> asStringID(final Iterable<byte[]> in) {

		return Iterables.transform(in, new Function<byte[], String>() {

			@Override

			public String apply(final byte[] input) {

				return getID(new String(input));

			}

		});

	}

	private Map<String, Set<EntityInfo>> getRelMap(final Context context,

			final String prefixFilter,

			final String entity,

			final String columnFamily,

			final Set<String> filter) throws IOException {

		System.out.println(String.format("loading %s, %s", entity, columnFamily));

		final Map<String, Set<EntityInfo>> out = Maps.newHashMap();

		final String tableName = context.getConfiguration().get("hbase.mapred.inputtable");

		System.out.println(String.format("table name: '%s'", tableName));

		try (final HTable table = new HTable(context.getConfiguration(), tableName);

				final ResultScanner res = scanTable(table, prefixFilter, entity, columnFamily)) {

			for (final Result r : res) {

				final byte[] bodyB = r.getValue(Bytes.toBytes(entity), Bytes.toBytes("body"));

				if (bodyB == null) {

					context.getCounter("missing body", entity).increment(1);

				} else {

					final OafEntity oafEntity = OafDecoder.decode(bodyB).getEntity();

					final EntityInfo kv = getEntityInfo(oafEntity, filter);

					if (kv != null) {

						final Map<byte[], byte[]> relMap = r.getFamilyMap(Bytes.toBytes(columnFamily));

						if (MapUtils.isNotEmpty(relMap)) {

							for (final String id : asStringID(relMap.keySet())) {

								if (!out.containsKey(id)) {

									out.put(id, new HashSet<EntityInfo>());

								}

								out.get(id).add(kv);

							}

						} else {

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

						}

					}

				}

			}

		}

		System.out.println(String.format("loaded map for %s, %s, size: %s", entity, columnFamily, out.size()));

		return out;

	}

	private EntityInfo getEntityInfo(final OafEntity entity, final Set<String> filter) {

		final String id = getID(entity.getId());

		switch (entity.getType()) {

		case datasource:

			final String dsType = entity.getDatasource().getMetadata().getDatasourcetype().getClassid();

			if (!filter.contains(dsType)) { return null; }

			return new EntityInfo(id, entity.getDatasource().getMetadata().getOfficialname().getValue());

		case organization:

			if (filter.contains(prefix(id))) { return null; }

			return new EntityInfo(id, entity.getOrganization().getMetadata().getLegalname().getValue());

		default:

			throw new IllegalArgumentException("invalid entity: " + entity);

		}

	}

	private ResultScanner scanTable(final HTable table, final String prefixFilter, final String entity, final String columnFamily) throws IOException {

		final Scan scan = new Scan();

		final FilterList fl = new FilterList(Operator.MUST_PASS_ALL);

		fl.addFilter(new PrefixFilter(Bytes.toBytes(prefixFilter)));

		scan.setFilter(fl);

		scan.addFamily(Bytes.toBytes(entity));

		scan.addFamily(Bytes.toBytes(columnFamily));

		return table.getScanner(scan);

	}

	private String getID(final String s) {

		return StringUtils.substringAfter(s, "|");

	}

	private String prefix(final String s) {

		return StringUtils.substringBefore(s, "::");

	}

}