D-Net

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

import java.io.IOException;

import com.googlecode.protobuf.format.JsonFormat;

import eu.dnetlib.data.graph.model.DNGFRowKeyDecoder;

import eu.dnetlib.data.mapreduce.hbase.dli.key.DliKey;

import eu.dnetlib.data.mapreduce.hbase.dli.key.DliKey.KeyType;

import eu.dnetlib.data.proto.DNGFProtos.DNGF;

import eu.dnetlib.data.proto.dli.ScholixObjectProtos.Scholix;

import eu.dnetlib.dli.proto.DngfToScholixConverter;

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 static eu.dnetlib.data.mapreduce.util.dao.HBaseTableDAO.getMetadata;

import static eu.dnetlib.data.mapreduce.util.dao.HBaseTableDAO.rel;

/**

 * Created by sandro on 2/15/17.

 */

public class PrepareScholixDataMapper extends TableMapper<DliKey, Text> {

    private DliKey outKey;

    private Text outValue;

    @Override

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

        outKey = new DliKey();

        outValue = new Text();

    }

    @Override

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

        final DNGF entity = getMetadata(value, DNGFRowKeyDecoder.decode(key.copyBytes()).getType());

        if (isValid(entity) && !deletedByInference(entity)) {

            final Scholix.Builder source = DngfToScholixConverter.withSource(entity.getEntity());

            emit(KeyType.entity, entity.getEntity().getId(), context, JsonFormat.printToString(source.build()));

            rel(value, "isMergedIn", "merges", "isSimilarTo").values().forEach(r -> {

                if (!deletedByInference(r)) {

                    final Scholix.Builder target = DngfToScholixConverter.withTarget(entity.getEntity(), r.getRel());

                    emit(KeyType.rel, r.getRel().getTarget(), context, JsonFormat.printToString(target.build()));

                }

            });

        }

    }

    private boolean isValid(final DNGF entity) {

        return (entity != null) && entity.isInitialized();

    }

    private boolean deletedByInference(final DNGF oaf) {

        return oaf.getDataInfo().getDeletedbyinference();

    }

    private void emit(final DliKey.KeyType keyType, final String id, final Context context, final String data) {

        outKey.setKeyType(keyType);

        outKey.setId(id);

        outValue.set(data.getBytes());

        try {

            context.write(outKey, outValue);

        } catch (Exception e) {

            e.printStackTrace();

            throw new RuntimeException(e);

        }

    }

}