dnet-hadoop/dhp-workflows/dhp-enrichment/src/main/java/eu/dnetlib/dhp/bulktag/eosc/SparkEoscBulkTag.java

package eu.dnetlib.dhp.bulktag.eosc;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.gson.Gson;
import eu.dnetlib.dhp.application.ArgumentApplicationParser;
import eu.dnetlib.dhp.bulktag.SparkBulkTagJob;
import eu.dnetlib.dhp.bulktag.community.*;
import eu.dnetlib.dhp.schema.common.ModelSupport;
import eu.dnetlib.dhp.schema.oaf.Context;
import eu.dnetlib.dhp.schema.oaf.DataInfo;
import eu.dnetlib.dhp.schema.oaf.Qualifier;
import eu.dnetlib.dhp.schema.oaf.Result;
import eu.dnetlib.dhp.schema.oaf.utils.OafMapperUtils;
import org.apache.commons.io.IOUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.print.attribute.DocAttributeSet;
import java.io.Serializable;
import java.util.*;
import java.util.stream.Collectors;

import static eu.dnetlib.dhp.PropagationConstant.removeOutputDir;
import static eu.dnetlib.dhp.bulktag.community.TaggingConstants.*;
import static eu.dnetlib.dhp.bulktag.community.TaggingConstants.TAGGING_TRUST;
import static eu.dnetlib.dhp.common.SparkSessionSupport.runWithSparkSession;
import static eu.dnetlib.dhp.schema.common.ModelConstants.DNET_PROVENANCE_ACTIONS;

/**
 * @author miriam.baglioni
 * @Date 21/07/22
 */
public class SparkEoscBulkTag implements Serializable {

    private static final Logger log = LoggerFactory.getLogger(SparkEoscBulkTag.class);
    public static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();

    public static void main(String[] args) throws Exception {
        String jsonConfiguration = IOUtils
                .toString(
                        SparkEoscBulkTag.class
                                .getResourceAsStream(
                                        "/eu/dnetlib/dhp/bulktag/input_bulkTag_parameters.json"));

        final ArgumentApplicationParser parser = new ArgumentApplicationParser(jsonConfiguration);
        parser.parseArgument(args);

        Boolean isSparkSessionManaged = Optional
                .ofNullable(parser.get("isSparkSessionManaged"))
                .map(Boolean::valueOf)
                .orElse(Boolean.TRUE);
        log.info("isSparkSessionManaged: {}", isSparkSessionManaged);

        final String inputPath = parser.get("sourcePath");
        log.info("inputPath: {}", inputPath);

        final String outputPath = parser.get("outputPath");
        log.info("outputPath: {}", outputPath);

        String datasourceMapPath = parser.get("datasourceMapPath");
        log.info("datasourceMapPath: {}", datasourceMapPath);

        final String resultClassName = parser.get("resultTableName");
        log.info("resultTableName: {}", resultClassName);

        Class<? extends Result> resultClazz = (Class<? extends Result>) Class.forName(resultClassName);

        SparkConf conf = new SparkConf();
        CommunityConfiguration cc;

        runWithSparkSession(
                conf,
                isSparkSessionManaged,
                spark -> {
                    removeOutputDir(spark, outputPath);
                    execBulkTag(spark, inputPath, outputPath, datasourceMapPath, resultClazz);
                });
    }

    private static <R extends Result> void execBulkTag(
            SparkSession spark,
            String inputPath,
            String outputPath,
            String datasourceMapPath,
            Class<R> resultClazz) {

        final List<String> hostedByList = Arrays.asList(readPath(spark, datasourceMapPath, DatasourceMaster.class)
                .map((MapFunction<DatasourceMaster, String>) dm -> dm.getMaster(), Encoders.STRING())
                .collect());

        readPath(spark, inputPath, resultClazz)
                .map(patchResult(), Encoders.bean(resultClazz))
                .filter(Objects::nonNull)
                .map(
                        (MapFunction<R, R>) value -> enrich(value, hostedByList),
                        Encoders.bean(resultClazz))
                .write()
                .mode(SaveMode.Overwrite)
                .option("compression", "gzip")
                .json(outputPath);
    }

    private static <R extends Result> R enrich(R value, List<String> hostedByList) {
        if(value.getInstance().stream().anyMatch(i -> hostedByList.contains(i.getHostedby().getKey())) ||
                (value.getEoscifguidelines() != null && value.getEoscifguidelines().size() > 0)){
            Context context = new Context();
            context.setId("eosc");
            OafMapperUtils.dataInfo(false, BULKTAG_DATA_INFO_TYPE,true,false,
                    OafMapperUtils.qualifier(CLASS_ID_DATASOURCE, CLASS_NAME_BULKTAG_DATASOURCE,
                            DNET_PROVENANCE_ACTIONS, DNET_PROVENANCE_ACTIONS), TAGGING_TRUST);
            value.getContext().add(context);

        }
        return value;

    }

    public static <R> Dataset<R> readPath(
            SparkSession spark, String inputPath, Class<R> clazz) {
        return spark
                .read()
                .textFile(inputPath)
                .map((MapFunction<String, R>) value -> OBJECT_MAPPER.readValue(value, clazz), Encoders.bean(clazz));
    }

    // TODO remove this hack as soon as the values fixed by this method will be provided as NON null
    private static <R extends Result> MapFunction<R, R> patchResult() {
        return r -> {
            if (r.getDataInfo().getDeletedbyinference() == null) {
                r.getDataInfo().setDeletedbyinference(false);
            }
            if (r.getContext() == null) {
                r.setContext(new ArrayList<>());
            }
            return r;
        };
    }

}