D-Net

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

import com.google.common.base.Splitter;

import com.google.common.collect.Lists;

import com.google.common.collect.Sets;

import com.googlecode.protobuf.format.JsonFormat;

import eu.dnetlib.data.mapreduce.util.DedupUtils;

import eu.dnetlib.data.proto.OafProtos;

import eu.dnetlib.data.proto.TypeProtos;

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.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

import java.util.Map;

import java.util.Set;

public class ConfigurableExportMapper extends TableMapper<Text, Text> {

    private static final String CF_DEFAULT = "result";

    private static final String QUALIFIER_DEFAULT = "body";

    private Text keyOut;

    private Text valueOut;

    private Set<String> columnFamilies = Sets.newHashSet();

    private Set<String> qualifiers = Sets.newHashSet();

    enum ExportMode {

        ALL,

        DUPS_ONLY

    }

    private ExportMode exportMode = ExportMode.ALL;

    @Override

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

        keyOut = new Text("");

        valueOut = new Text();

        columnFamilies.addAll(Lists.newArrayList(Splitter.on(",").split(context.getConfiguration().get("column.families", CF_DEFAULT))));

        qualifiers.addAll(Lists.newArrayList(Splitter.on(",").split(context.getConfiguration().get("qualifiers", QUALIFIER_DEFAULT))));

        exportMode = ExportMode.valueOf(context.getConfiguration().get("export.mode", ExportMode.ALL.toString()));

    }

    @Override

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

        for(String cf : columnFamilies) {

            switch (exportMode) {

                case ALL:

                    export(value, context, cf);

                    break;

                case DUPS_ONLY:

                    final Map<byte[], byte[]> mergedInMap = value.getFamilyMap(DedupUtils.getDedupCF_mergedInBytes(cf));

                    if (mergedInMap != null && !mergedInMap.isEmpty()) {

                        export(value, context, cf);

                    }

                    break;

            }

        }

    }

    private void export(Result value, Context context, String cf) throws IOException, InterruptedException {

        for(String q : qualifiers) {

            final byte[] data = value.getValue(cf.getBytes(), q.getBytes());

            if (data != null) {

                OafProtos.Oaf oaf = OafProtos.Oaf.parseFrom(data);

                valueOut.set(JsonFormat.printToString(oaf));

                context.write(keyOut, valueOut);

                context.getCounter(cf, q).increment(1);

            }

        }

    }

}