D-Net

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

import java.io.IOException;

import eu.dnetlib.actionmanager.actions.AtomicAction;

import eu.dnetlib.data.mapreduce.JobParams;

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

import org.apache.hadoop.hbase.io.ImmutableBytesWritable;

import org.apache.hadoop.hbase.util.Bytes;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

/**

 * Created by claudio on 11/04/16.

 */

public class PromoteActionSetFromHDFSMapper extends Mapper<Text, Text, ImmutableBytesWritable, Put> {

	private ImmutableBytesWritable keyOut;

	@Override

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

		super.setup(context);

		keyOut = new ImmutableBytesWritable();

	}

	@Override

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

		final AtomicAction aa = AtomicAction.fromJSON(value.toString());

		final byte[] keyOutBytes = Bytes.toBytes(aa.getTargetRowKey());

		keyOut.set(keyOutBytes);

		final Put put = new Put(keyOutBytes);

		put.setWriteToWAL(JobParams.WRITE_TO_WAL);

		put.add(Bytes.toBytes(aa.getTargetColumnFamily()), Bytes.toBytes(aa.getTargetColumn()), aa.getTargetValue());

		context.write(keyOut, put);

	}

}