D-Net

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

import com.google.protobuf.InvalidProtocolBufferException;

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.LinkDescriptor;

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

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

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

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

import eu.dnetlib.data.proto.RelTypeProtos.SubRelType;

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

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

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 org.apache.log4j.Logger;

import java.io.IOException;

/**

 * export

	private static final String LAST_EXECUTION_DATE_PATTERN = "yyyy-MM-dd";

	private Logger log = Logger.getLogger(this.getClass());

	private EntityConfigTable entityConfigTable;

	private String lastExecutionDate = "";

		RESULT, PROJECT, DATASOURCE, PERSON, ORGANIZATION, RESULT_NO_DATE, PROJECT_NO_DATE, DATASOURCE_NO_DATE,

		PERSON_NO_DATE, RGANIZATION_NO_DATE, DELETED_BY_INFERENCE, NOT_DELETED_BY_INFERENCE, TOTAL_ENTITIES,

		TOTAL_RELATIONS, UPDATED, NOT_UPDATED, NO_DATE

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

		loadEntityConfig(context);

		DELIM = context.getConfiguration().get("lod.delim");

		lastExecutionDate = context.getConfiguration().get("lod.lastExecutionDate");

	protected void map(final ImmutableBytesWritable keyIn, final Result result, final Context context)

			throws IOException {

		final Type type = decoder.getType();

		final Oaf oaf = UpdateMerger.mergeBodyUpdates(context, result.getFamilyMap(Bytes.toBytes(type.toString())));

				context.getCounter(ENTITIES_COUNTER.DELETED_BY_INFERENCE).increment(1);

			} else {

				context.getCounter(ENTITIES_COUNTER.NOT_DELETED_BY_INFERENCE).increment(1);

			}

			try {

				emitProtos(context, result, oaf);

			} catch (ParseException e) {

				// TODO Auto-generated catch block

				e.printStackTrace();

			}

		}

		try {

			if (oaf != null && oaf.isInitialized()) {

				return true;

			}

			log.error("invalid proto", e);

		}

	}

		String dateOfTransformationString = "";

		try {

			SimpleDateFormat simpleDateFormat = new SimpleDateFormat(DATE_OF_TRANSFORMATION_PATTERN,

					Locale.getDefault());

			simpleDateFormat.setTimeZone(TimeZone.getTimeZone("UTC"));

			dateOfTransformationString = oaf.getEntity().getDateoftransformation();

			if (dateOfTransformationString == null || dateOfTransformationString.isEmpty()

					|| dateOfTransformationString.equals(" ")) {

				context.getCounter(ENTITIES_COUNTER.NO_DATE).increment(1);

				return true;

			}

			simpleDateFormat.setTimeZone(TimeZone.getTimeZone("UTC"));

			Date lastExecDate = lastExecDateFormatter.parse(lastExecutionDate);

				context.getCounter(ENTITIES_COUNTER.UPDATED).increment(1);

				return true;

			}

		} catch (ParseException pe) {

			SimpleDateFormat simpleDateFormat = new SimpleDateFormat(LAST_EXECUTION_DATE_PATTERN, Locale.getDefault());

			simpleDateFormat.setTimeZone(TimeZone.getTimeZone("UTC"));

			dateOfTransformationString = oaf.getEntity().getDateoftransformation();

			if (dateOfTransformationString == null || dateOfTransformationString.isEmpty()

					|| dateOfTransformationString.equals(" ")) {

				context.getCounter(ENTITIES_COUNTER.NO_DATE).increment(1);

				return true;

			}

			SimpleDateFormat lastExecDateFormatter = new SimpleDateFormat(LAST_EXECUTION_DATE_PATTERN);

			simpleDateFormat.setTimeZone(TimeZone.getTimeZone("UTC"));

			Date lastExecDate = lastExecDateFormatter.parse(lastExecutionDate);

				context.getCounter(ENTITIES_COUNTER.UPDATED).increment(1);

				return true;

			}

		} catch (Exception e) {

			log.error("invalid date " + dateOfTransformationString, e);

			throw new IOException(e);

		}

		context.getCounter(ENTITIES_COUNTER.NOT_UPDATED).increment(1);

		return false;

	}

		Oaf.Builder oafBuilder = Oaf.newBuilder().setKind(oaf.getKind()).setDataInfo(oaf.getDataInfo());

		Type type = oaf.getEntity().getType();

		oafBuilder.setEntity(oaf.getEntity());

		// emit relation first so we can cache them to entity protos

		emitRelation(context, result, oaf, type, oafBuilder);

			emitEntity(context, oaf, oafBuilder);

		}

	}

		String serialized = Serializer.serialize(oafBuilder.build(), DELIM);

				context.getCounter(ENTITIES_COUNTER.valueOf(oaf.getEntity().getType().toString().toUpperCase()))

						.increment(1);

				// output entity types so we can have seperate files for each type

				Text TextKeyOut = new Text(oaf.getEntity().getType().toString());

				context.write((TextKeyOut), new Text(serialized));

					context.getCounter(

							ENTITIES_COUNTER.valueOf(oaf.getEntity().getType().toString().toUpperCase() + "_NO_DATE"))

							.increment(1);

//                    context.getCounter(ENTITIES_COUNTER.UPDATED).increment(1);

				}

//                else {

//                    context.getCounter(ENTITIES_COUNTER.NOT_UPDATED).increment(1);

//                }

				log.error("Error writing entity to M/R output", e);

			}

		}

	private void emitRelation(Context context, Result result, Oaf oaf, Type type, Oaf.Builder builder) {

		// derived relations; ResultLanguages, resultConcepts etc are

		// created here

			// Existing Hbase relations are generated here

			if (entityConfigTable.getDescriptors(type) != null && !entityConfigTable.getDescriptors(type).isEmpty()) {

				for (LinkDescriptor ld : entityConfigTable.getDescriptors(type)) {

					try {

								.getFamilyMap(Bytes.toBytes(ld.getRelDescriptor().getIt()));

							builder.getEntityBuilder().addCachedRel(rel);

							try {

								// keep all relations to one file

								Text TextKeyOut = new Text("relations");

									context.write((TextKeyOut), new Text(buff));

									context.getCounter(ENTITIES_COUNTER.TOTAL_RELATIONS).increment(1);

								}

								log.error("Error while writing Relation Proto to M/R output", e);

							}

					} catch (Exception e) {

						log.error("Error while decoding Relation Proto from HBase Body", e);

			}

		}

			try {

					if (!rel.contains("dedup")) {

						context.write((TextKeyOut), new Text(rel));

						context.getCounter(ENTITIES_COUNTER.TOTAL_RELATIONS).increment(1);

				} else {

					// for dedup entities write only dedup relationships: all the permutations

					// of children

					if (rel.contains("dedup")) {

						context.write((TextKeyOut), new Text(rel));

						context.getCounter(ENTITIES_COUNTER.TOTAL_RELATIONS).increment(1);

					}

				log.error("Error writing relations to output : " + rel);

			}

		}

		relationsList.clear();

	}

			throws IOException, InterruptedException {

				if (isValid(decodedOaf)) {

					OafRel.Builder relBuilder = OafRel.newBuilder(decodedOaf.getRel());

							&& isDedupSelf(relBuilder)) {

						// log.error("invalid protto", e);

						OafRel oafRel = relBuilder.setRelType(ld.getRelDescriptor().getRelType()).build();

						relBuilder.setCachedTarget(body.getEntity());

						rels.add(oafRel);

					}

				}

		}

	}

		try {

			return Oaf.parseFrom(body);

		} catch (InvalidProtocolBufferException e) {

			log.error("Invalid Protos", e);

		}

		return null;

	}

		String indexConf = context.getConfiguration().get("index.conf");

			log.error("NULL ENTITY CONFIGURATION TABLE IN MAPPER  : ");

			throw new InterruptedException("Null enitity configuration in mappper");

		return rel.getSource().contains(rel.getTarget());

	}

		return columnMap != null && !columnMap.isEmpty();

	}

		return oaf.getDataInfo().getDeletedbyinference();

	}

	protected void cleanup(Context context) throws IOException, InterruptedException {

		super.cleanup(context);

	}

		return entityConfigTable;

	}

		this.entityConfigTable = entityConfigTable;

	}