D-Net

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

import eu.dnetlib.data.proto.DNGFProtos;

import eu.dnetlib.data.proto.FieldTypeProtos;

import eu.dnetlib.data.proto.WdsDatasetProtos;

import eu.dnetlib.data.proto.WdsPublicationProtos;

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

import org.apache.hadoop.hbase.mapreduce.TableReducer;

import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

import java.util.ArrayList;

import java.util.List;

public class ExtendProjectInfoReducer extends TableReducer<ImmutableBytesWritable, ImmutableBytesWritable, ImmutableBytesWritable> {

    private ImmutableBytesWritable outKey = new ImmutableBytesWritable();

    private ImmutableBytesWritable outValue = new ImmutableBytesWritable();

    @Override

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

    }

    @Override

    protected void reduce(ImmutableBytesWritable key, Iterable<ImmutableBytesWritable> values, Context context) throws IOException, InterruptedException {

        List<FieldTypeProtos.ProjectRelation> projects = new ArrayList<>();

        final DNGFProtos.DNGFEntity.Builder dataset = DNGFProtos.DNGFEntity.newBuilder() ;

        values.forEach(it ->{

            final DNGFProtos.DNGFEntity entity = tryToParseEntity(it);

            if (entity != null)

                dataset.mergeFrom(entity);

            else {

                FieldTypeProtos.ProjectRelation p = tryToParseProject(it);

                if (p!= null) projects.add(p);

            }

        });

        if (dataset.hasDataset() && projects.size() > 0 ) {

            projects.forEach(p -> dataset.getDatasetBuilder().getMetadataBuilder().addExtension(WdsDatasetProtos.WdsDataset.projects,p));

            context.getCounter("PropagatedProjects", "for Dataset").increment(projects.size());

            outKey.set(key.copyBytes());

            outValue.set(dataset.build().toByteArray());

            context.write(outKey, outValue);

        }

    }

    private DNGFProtos.DNGFEntity tryToParseEntity(final ImmutableBytesWritable input) {

        try {

            DNGFProtos.DNGFEntity dngfEntity = DNGFProtos.DNGFEntity.parseFrom(input.copyBytes());

            return dngfEntity;

        } catch (Throwable e) {

            return null;

        }

    }

    private FieldTypeProtos.ProjectRelation tryToParseProject(final ImmutableBytesWritable input) {

        try {

            FieldTypeProtos.ProjectRelation  project = FieldTypeProtos.ProjectRelation .parseFrom(input.copyBytes());

            return project;

        } catch (Throwable e) {

            return null;

        }

    }

}