D-Net

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

import java.io.IOException;

import java.util.Set;

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

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 com.google.common.base.Splitter;

import com.google.common.collect.Sets;

public class PromoteActionsMapper extends TableMapper<ImmutableBytesWritable, Put> {

	private static final String LATEST_RAW_SETS = "latestRawSets";

	private Set<byte[]> latestRawSets = Sets.newHashSet();

	@Override

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

		final String s = context.getConfiguration().get(LATEST_RAW_SETS);

		if (s != null) {

			for (String set : Sets.newHashSet(Splitter.on(",").omitEmptyStrings().trimResults().split(s))) {

				latestRawSets.add(Bytes.toBytes(set));

			}

		}

		if (latestRawSets.isEmpty()) { throw new IOException("Input parameter (" + LATEST_RAW_SETS + ") is missing or empty: " + s); }

	}

	@Override

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

		for (byte[] s1 : value.getFamilyMap(Bytes.toBytes("set")).keySet()) {

			for (byte[] s2 : latestRawSets) {

				if (Bytes.equals(s1, s2)) {

					promoteAction(value, context);

				}

			}

		}

	}

	private void promoteAction(final Result value, final Context context) throws IOException, InterruptedException {

		final byte[] cf = Bytes.toBytes("target");

		final byte[] tkey = value.getValue(cf, Bytes.toBytes("rowKey"));

		final byte[] tcf = value.getValue(cf, Bytes.toBytes("columnFamily"));

		final byte[] tc = value.getValue(cf, Bytes.toBytes("column"));

		final byte[] tv = value.getValue(cf, Bytes.toBytes("content"));

		final Put put = new Put(tkey);

		put.add(tcf, tc, tv);

		context.getCounter("Actions", Bytes.toString(tcf)).increment(1);

		context.write(new ImmutableBytesWritable(tkey), put);

	}

}