D-Net

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

import com.google.common.base.Splitter;

import com.google.common.collect.Sets;

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

import eu.dnetlib.data.proto.*;

import org.apache.commons.collections.MapUtils;

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 java.io.IOException;

import java.util.Map;

import java.util.Set;

import java.util.stream.Collectors;

import static eu.dnetlib.data.mapreduce.hbase.propagation.PropagationConstants.*;

/**

 * Created by miriam on 17/08/2018.

 */

public class PropagationCountryFromDsOrgResultMapper extends TableMapper<Key, ImmutableBytesWritable> {

    private ImmutableBytesWritable valueOut;

    private Set<String> datasourceTypes = Sets.newHashSet("pubsrepository::institutional");

    @Override

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

        super.setup(context);

        valueOut = new ImmutableBytesWritable();

        datasourceTypes.addAll(Splitter.on(",").omitEmptyStrings().splitToList(context.getConfiguration().get("datasource.types", "")));

    }

    @Override

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

        final Map<byte[], byte[]> resultMap = value.getFamilyMap(Bytes.toBytes("result"));

        final byte[] body = resultMap.get(Bytes.toBytes("body"));

        if (body != null){

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

            OafProtos.OafEntity entity = oaf.getEntity();

            switch (OafRowKeyDecoder.decode(keyIn.copyBytes()).getType()) {

                case datasource:

                    DatasourceProtos.Datasource datasource = entity.getDatasource();

                    if (datasource == null){

                        throw new RuntimeException("oaf type is datasource, but datasource proto is not found in oafproto");

                    }

                    final Key datasource1 = Key.datasource(entity.getId());

                    String dsType = datasource.getMetadata().getDatasourcetype().getClassid();

                    if(datasourceTypes.contains(dsType)) {

                        context.getCounter(COUNTER_PROPAGATION, dsType).increment(1);

                        valueOut.set(ONE.getBytes());

                        context.write(datasource1,valueOut);

                    }else{

                        context.getCounter(COUNTER_PROPAGATION, String.format("not %s", dsType)).increment(1);

                        valueOut.set(ZERO.getBytes());

                        context.write(datasource1,valueOut);

                    }

                    break;

                case organization:

                    OrganizationProtos.Organization organization = entity.getOrganization();

                    if (organization == null){

                        throw new RuntimeException("oaf type is organization, but organization proto is not found in oafproto");

                    }

                    FieldTypeProtos.Qualifier country = organization.getMetadata().getCountry();

                    if (country == null){

                        context.getCounter(COUNTER_PROPAGATION,"country elem does not exists").increment(1);

                    }else{

                        final Map<byte[], byte[]> ds_org = value.getFamilyMap("datasourceOrganization_provision_isProvidedBy".getBytes());

                        if (MapUtils.isNotEmpty(ds_org)) {

                            for(String dsId : ds_org.keySet().stream().map(String::new).collect(Collectors.toList())) {

                                valueOut.set(country.getClassid().getBytes());

                                context.write(Key.organization(dsId), valueOut);

                                context.getCounter(COUNTER_PROPAGATION,"country ").increment(1);

                            }

                        }

                    }

                    break;

                case result:

                    ResultProtos.Result result = entity.getResult();

                    for(ResultProtos.Result.Instance instance : result.getInstanceList()) {

                        //todo add check if key is not empty and field is not null

                        String hostedBy = instance.getHostedby().getKey();

                        valueOut.set(entity.getId().getBytes());

                        context.write(Key.publication(hostedBy), valueOut);

                        String collectedFrom = instance.getCollectedfrom().getKey();

                        if (!hostedBy.equals(collectedFrom)) {

                            context.write(Key.publication(collectedFrom),valueOut);

                        }

                    }

                    break;

            }

        }

    }

}