D-Net

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

import java.io.IOException;

import java.util.HashSet;

import java.util.Map;

import java.util.Set;

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.data.broker.model.openaire.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;

import org.apache.commons.collections.MapUtils;

import org.apache.commons.lang.StringUtils;

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 static eu.dnetlib.data.mapreduce.hbase.broker.mapping.EventFactory.asEvent;

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

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

/**

 * 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 {

		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 (String projectId : asStringID(resultProject.keySet())) {

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

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

					for (EntityInfo organization : organizations) {

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

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

							for (EntityInfo datasource : datasources) {

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

									//emit event for datasourceId

									final EventMessage event = asEvent(oaf.getEntity(), Topic.ADD_PROJECT, datasource.getId(), datasource.getName());

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

									//event.setPayload(HighlightFactory.highlightEnrichPid(payload, Lists.newArrayList(pids)).toJSON());

									event.setPayload(payload.toJSON());

									emit(event, context);

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

								}

							}

						}

					}

				}

			}

		}

	}

	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 ResultScanner res = scanTable(context, prefixFilter, entity, columnFamily);

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

					}

				}

			}

		}

		res.close();

		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 Context context, 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));

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

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

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

		return table.getScanner(scan);

	}

	private String getID(final String s) {

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

	}

	private String prefix(final String s) {

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

	}

}