D-Net

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

import java.io.IOException;

import java.util.Set;

import org.apache.hadoop.hbase.KeyValue;

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

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;

import eu.dnetlib.data.mapreduce.JobParams;

public class DeleteRecordsMapper extends TableMapper<ImmutableBytesWritable, Delete> {

	private Set<String> delPrefix;

	@Override

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

		super.setup(context);

		delPrefix = extractPrefixes(context.getConfiguration().get("nsPrefix"));

		String table = context.getConfiguration().get(JobParams.HBASE_SOURCE_TABLE).trim();

		System.out.println("I start to delete records (table=" + table + ", prefixes=" + delPrefix);

	}

	@Override

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

		byte[] bkey = key.get();

		System.out.println("EVALUATING " + Bytes.toString(bkey));

		if (testValue(bkey)) { // If key contains repoId I delete the row

			context.write(key, new Delete(bkey));

			System.out.println("   --- DELETED");

			context.getCounter("MDStore Cleaner", "Deleted records").increment(1);

		} else { // otherwise I try to delete the columns containing repoId 

			for (KeyValue kv : value.list()) {

				byte[] col = kv.getQualifier();

				if (testValue(col)) {

					Delete d = new Delete(bkey);

					d.deleteColumns(kv.getFamily(), col);

					context.write(key, d);

					System.out.println("   --- DELETED COLUMN: " + Bytes.toString(col));

					context.getCounter("MDStore Cleaner", "Deleted columns").increment(1);

				}

			}

		}

	}

	private Set<String> extractPrefixes(String s) {

		return Sets.newHashSet(Splitter.on(",").trimResults().omitEmptyStrings().split(s));

	}

	private boolean testValue(byte[] bkey) {

		String s = Bytes.toString(bkey);

		if (s.contains("|") && s.contains("::")) {

			return delPrefix.contains(s.substring(s.indexOf("|") + 1, s.indexOf("::")));

		} else if (s.contains("::")) {

			return delPrefix.contains(s.substring(0, s.indexOf("::")));

		} else {

			return false;

		}

	}

}