D-Net

package eu.dnetlib.data.mapreduce.es;

import java.io.IOException;

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

import org.apache.hadoop.io.NullWritable;

import com.google.protobuf.InvalidProtocolBufferException;

import com.googlecode.protobuf.format.JsonFormat;

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

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

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

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

public class ElasticsearchFeedMapper extends TableMapper<NullWritable, BytesWritable> {

	@Override

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

	}

	@Override

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

		final OafRowKeyDecoder keyDecoder = OafRowKeyDecoder.decode(keyIn.copyBytes());

		final Oaf oaf = mergeUpdates(value, context, keyDecoder.getType(), keyDecoder);

		if (isValid(oaf)) {

			context.write(NullWritable.get(), new BytesWritable(Bytes.toBytes(JsonFormat.printToString(oaf))));

		}

	}

	private Oaf mergeUpdates(final Result value, final Context context, final Type type, final OafRowKeyDecoder keyDecoder)

			throws InvalidProtocolBufferException {

		try {

			return UpdateMerger.mergeBodyUpdates(context, value.getFamilyMap(Bytes.toBytes(type.toString())));

		} catch (final InvalidProtocolBufferException e) {

			System.err.println(String.format("Unable to parse proto (Type: %s) in row: %s", type.toString(), keyDecoder.getKey()));

			throw e;

		}

	}

	private boolean isValid(final Oaf oaf) {

		return (oaf != null) && oaf.isInitialized();

	}

}

<!-- 		<dependency> -->

<!-- 		    <groupId>org.elasticsearch</groupId> -->

<!-- 		    <artifactId>elasticsearch-hadoop</artifactId> -->

<!-- 		    <version>2.0.2</version> -->

<!-- 		</dependency>		 -->

		<dependency>

		    <groupId>org.elasticsearch</groupId>

		    <artifactId>elasticsearch-hadoop-mr</artifactId>

		    <version>2.0.2</version>

		</dependency>