D-Net

package eu.dnetlib.data.mapreduce.hbase.lodExport.preprocessing;

import eu.dnetlib.data.mapreduce.hbase.lodExport.utils.LodConfiguration;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

import org.apache.log4j.Logger;

import java.io.IOException;

/**

 * Mapper Class that reads HBASE contents and prepares them for the StatsDB

 * export

 */

public class TargetMapper extends Mapper<LongWritable, Text, Text, Text> {

    private Logger log = Logger.getLogger(this.getClass());

    private String delim;

    private LodConfiguration lodConfiguration;

    public static enum ENTITIES_COUNTER {

        TARGET_ENTITIES,

    }

    @Override

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

        lodConfiguration = new LodConfiguration();

        lodConfiguration.load(context.getConfiguration().get("lod.sourceMappings"));

    }

    @Override

    protected void map(final LongWritable keyIn, final Text result, final Context context) throws IOException {

        try {

            //get ID - output source_recordID so we can group by id and get all props of a record

            StringBuffer value = new StringBuffer();

            //TODO remove this and append <,> to all fields in mappings

            String[] Fields = result.toString().replace("<", "").replace(">", "").split("\t");

            //here addd all  fields as array props and append "\t"

            // betweeen them so we can write directly to output

            //DO NOT enter id- we'll get it from key output

            int i = 1;

            //extract entity type from subject

            String[] tmp = Fields[0].split("/");

            String type = tmp[4];

            String subject = Fields[0];

            if (lodConfiguration.entityExists(type)) {

                while (i < Fields.length - 1) {

                    String field = Fields[i];

                    String fieldValue = Fields[i + 1];

                    // if (lodConfiguration.isValidField(type, field.replace("/", "\\/")))

                    {

                        value.append(field + "\t" + fieldValue).append(",");

                    }

                    i += 2;

                }

                // write out type,source_ID as key, and rest of props as value

                //key = person, <http://lod.openaire.eu/data/person/od______1108::ac127c39675469b355dc4f2aa4b301fd>

                Text key = new Text("TARGET" + "," + type + "," + subject);

                context.write(key, new Text(value.toString()));

                context.getCounter(SourceMapper.ENTITIES_COUNTER.SOURCE_ENTITIES).increment(1);

            }

        } catch (Exception e) {

            log.error("Error writing entity to M/R output", e);

        }

    }

    @Override

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

        super.cleanup(context);

    }

}