D-Net

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

import java.io.IOException;

import eu.dnetlib.data.mapreduce.JobParams;

import org.apache.commons.io.IOUtils;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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;

import com.googlecode.protobuf.format.JsonFormat;

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

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

	/**

	 * logger.

	 */

	private static final Log log = LogFactory.getLog(ImportInformationSpaceDumpMapper.class); // NOPMD by marko on 11/24/08 5:02 PM

	private static final String SEPARATOR = "@";

	private ImmutableBytesWritable ibw;

	@Override

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

		super.setup(context);

		ibw = new ImmutableBytesWritable();

	}

	@Override

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

		try {

			String[] split = key.toString().split(SEPARATOR);

			if (split.length != 3)

				throw new IllegalArgumentException("malformed key: " + key.toString());

			byte[] rowKey = Bytes.toBytes(split[0]);

			byte[] columnFamily = Bytes.toBytes(split[1]);

			byte[] qualifier = Bytes.toBytes(split[2]);

			final Put put = new Put(rowKey);

			put.setWriteToWAL(JobParams.WRITE_TO_WAL);

			ibw.set(rowKey);

			if (StringUtils.isNotBlank(value.toString())) {

				Oaf.Builder oaf = Oaf.newBuilder();

				JsonFormat.merge(value.toString(), oaf);

				put.add(columnFamily, qualifier, oaf.build().toByteArray());

			} else {

				put.add(columnFamily, qualifier, Bytes.toBytes(""));

			}

			context.write(ibw, put);

		} catch (final Throwable e) {

			log.error(String.format("error importing the following record on HBase: \nkey: '%s'\nvalue: %s", key.toString(), value.toString()), e);

			context.getCounter("error", e.getClass().getName()).increment(1);

			throw new RuntimeException(e);

		}

	}

}