D-Net

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

import java.io.IOException;

import java.util.List;

import java.util.NavigableMap;

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.http.HttpResponse;

import org.apache.http.client.HttpClient;

import org.apache.http.client.methods.HttpPost;

import org.apache.http.client.methods.HttpUriRequest;

import org.apache.http.impl.client.DefaultHttpClient;

import org.apache.http.params.BasicHttpParams;

import org.apache.http.params.HttpParams;

import com.google.common.collect.Lists;

import com.google.gson.Gson;

import eu.dnetlib.data.mapreduce.hbase.VolatileColumnFamily;

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

public class FindPersonCoauthorsMapper extends TableMapper<Text, Text> {

	private final HttpClient client = new DefaultHttpClient();

	private final String url = "http://146.48.87.97:8888/addData";

	@Override

	protected void setup(final Context context) {

		// url = context.getConfiguration().get("dedup.person.coauthors.service.url");

	}

	@Override

	protected void map(final ImmutableBytesWritable rowkey, final Result row, final Context context) throws IOException, InterruptedException {

		final NavigableMap<byte[], byte[]> candidates = row.getFamilyMap(Bytes.toBytes(VolatileColumnFamily.dedupPerson.toString()));

		if ((candidates == null) || candidates.isEmpty()) return;

		final List<String> coauthors = Lists.newArrayList();

		for (final byte[] b : row.getFamilyMap(Bytes.toBytes(RelType.personResult.toString())).keySet()) {

			coauthors.add(Bytes.toString(b));

		}

		for (final byte[] candidate : candidates.keySet()) {

			emit(context, Bytes.toString(candidate), coauthors);

		}

	}

	private void emit(final Context context, final String candidate, final List<String> coauthors) {

		try {

			final HttpUriRequest request = new HttpPost(url);

			final HttpParams params = new BasicHttpParams();

			params.setParameter("id", candidate);

			params.setParameter("data", (new Gson().toJson(coauthors)));

			request.setParams(params);

			final HttpResponse response = client.execute(request);

			context.getCounter("HTTP call", "code " + response.getStatusLine().getStatusCode()).increment(1);

		} catch (final Exception e) {

			context.getCounter("HTTP call", "Exception " + e.getClass()).increment(1);

		}

	}

}