Index: modules/dnet-openaire-lodexport/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/utils/NewSerializer.java
===================================================================
--- modules/dnet-openaire-lodexport/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/utils/NewSerializer.java	(revision 45698)
+++ modules/dnet-openaire-lodexport/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/utils/NewSerializer.java	(nonexistent)
@@ -1,734 +0,0 @@
-package eu.dnetlib.data.mapreduce.hbase.lodExport.utils;
-
-import eu.dnetlib.data.mapreduce.util.LicenseComparator;
-import eu.dnetlib.data.proto.DatasourceProtos.Datasource.Metadata;
-import eu.dnetlib.data.proto.FieldTypeProtos;
-import eu.dnetlib.data.proto.FieldTypeProtos.StringField;
-import eu.dnetlib.data.proto.FieldTypeProtos.StructuredProperty;
-import eu.dnetlib.data.proto.OafProtos.Oaf;
-import eu.dnetlib.data.proto.OafProtos.OafEntity;
-import eu.dnetlib.data.proto.OafProtos.OafRel;
-import eu.dnetlib.data.proto.OrganizationProtos.Organization;
-import eu.dnetlib.data.proto.PersonProtos;
-import eu.dnetlib.data.proto.ProjectProtos.Project;
-import eu.dnetlib.data.proto.ResultProtos.Result;
-import eu.dnetlib.data.proto.ResultProtos.Result.Instance;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Set;
-
-/**
- * @author eri Simple serializer that parses input Oaf Protos and prepares them
- *         for sqoop
- */
-public class NewSerializer {
-    private static final String TEXT_PATTERN = "[^a-zA-Z0-9 .-_:/@+=-]";
-    private static final String ID_PATTERN = "[^a-zA-Z0-9.:/_-]";
-    private static final String DOI_PATTERN = ".*\\/(10.)";
-
-    public static String Serialize(Oaf oaf, String DELIM) {
-        switch (oaf.getKind()) {
-            case entity:
-                OafEntity valueEntity = oaf.getEntity();
-                switch (valueEntity.getType()) {
-                    case datasource:
-                        return buildDatasource(valueEntity, DELIM);
-                    case organization:
-                        return buildOrganization(valueEntity, DELIM);
-                    case project:
-                        return buildProject(valueEntity, DELIM);
-                    case result:
-                        return buildResult(valueEntity, DELIM);
-                    case person:
-                        return buildPerson(valueEntity, DELIM);
-                    default:
-                        break;
-                }
-                break;
-            case relation:
-                OafRel valueRel = oaf.getRel();
-
-                return Serialize(valueRel, DELIM);
-
-        }
-
-        return null;
-
-    }
-
-
-    public static void extractRelations(Oaf oaf, String DELIM, Set<String> relations) {
-        OafEntity valueEntity = oaf.getEntity();
-        switch (valueEntity.getType()) {
-            case result:
-                getResultDatasources(valueEntity, DELIM, relations);
-                getDedups(valueEntity, DELIM, relations);
-            case datasource:
-                getDedups(valueEntity, DELIM, relations);
-            case person:
-                getDedups(valueEntity, DELIM, relations);
-            case organization:
-                getDedups(valueEntity, DELIM, relations);
-            default:
-        }
-
-    }
-
-    public static String Serialize(OafRel Rel, String DELIM) {
-        StringBuilder buff;
-        switch (Rel.getRelType()) {
-            case datasourceOrganization:
-                buff = new StringBuilder();
-                buff.append(Rel.getRelType().name()).append(DELIM).append("datasource").append(DELIM).append(cleanId(Rel.getSource())).append(DELIM)
-                        .append("organization").append(DELIM).append(cleanId(Rel.getTarget())).append(DELIM).append("\n");
-                return buff.toString();
-            case resultResult:
-                buff = new StringBuilder();
-                buff.append(Rel.getRelType().name()).append(DELIM).append("result").append(DELIM).append(cleanId(Rel.getSource())).append(DELIM)
-                        .append("result").append(DELIM).append(cleanId(Rel.getTarget())).append(DELIM);
-                return buff.toString();
-            case personPerson:
-                buff = new StringBuilder();
-                buff.append(Rel.getRelType().name()).append(DELIM).append("person").append(DELIM).append(cleanId(Rel.getSource())).append(DELIM)
-                        .append("person").append(DELIM).append(cleanId(Rel.getTarget())).append(DELIM);
-                return buff.toString();
-            case organizationOrganization:
-                buff = new StringBuilder();
-                buff.append(Rel.getRelType().name()).append(DELIM).append("organization").append(DELIM).append(cleanId(Rel.getSource())).append(DELIM)
-                        .append("organization").append(DELIM).append(cleanId(Rel.getTarget())).append(DELIM);
-                return buff.toString();
-            case personResult:
-                buff = new StringBuilder();
-                buff.append(Rel.getRelType().name()).append(DELIM).append("person").append(DELIM).append(cleanId(Rel.getSource())).append(DELIM)
-                        .append("result").append(DELIM).append(cleanId(Rel.getTarget())).append(DELIM);
-                return buff.toString();
-            case projectOrganization:
-                buff = new StringBuilder();
-                buff.append(Rel.getRelType().name()).append(DELIM).append("project").append(DELIM).append(cleanId(Rel.getSource())).append(DELIM)
-                        .append("organization").append(DELIM).append(cleanId(Rel.getTarget())).append(DELIM);
-                return buff.toString();
-            case projectPerson:
-                buff = new StringBuilder();
-                buff.append(Rel.getRelType().name()).append(DELIM).append("project").append(DELIM).append(cleanId(Rel.getSource())).append(DELIM)
-                        .append("person").append(DELIM).append(cleanId(Rel.getTarget())).append(DELIM);
-                return buff.toString();
-            case resultOrganization:
-                buff = new StringBuilder();
-                buff.append(Rel.getRelType().name()).append(DELIM).append("result").append(DELIM).append(cleanId(Rel.getSource())).append(DELIM)
-                        .append("organization").append(DELIM).append(cleanId(Rel.getTarget())).append(DELIM);
-                return buff.toString();
-//TODO - SOS RELATION RESULT PROJECT IS INVERTED! SOURCE IS PROJECT, TARGET IS RESULT
-
-            case resultProject:
-                buff = new StringBuilder();
-                buff.append(Rel.getRelType().name()).append(DELIM).append("result").append(DELIM).append(cleanId(Rel.getTarget())).append(DELIM)
-                        .append("project").append(DELIM).append(cleanId(Rel.getSource())).append(DELIM);
-                return buff.toString();
-            default:
-
-        }
-
-        return "";
-
-    }
-
-    private static String getHeader(OafEntity data, String DELIM) {
-        String SEPERATOR = ";";
-
-        StringBuilder buff = new StringBuilder();
-
-        //  EntityType
-        buff.append(data.getType().name()).append(DELIM);
-
-        // OpenaireID
-        buff.append(cleanId(data.getId())).append(DELIM);
-        //  dateOfTransformation
-
-        // TODO  CRITERIA FOR IDENTIFYING UPDATED RECORDS
-        buff.append(cleanId(data.getDateoftransformation())).append(DELIM);
-
-        //    dateOfCollection
-        buff.append(clean(data.getDateofcollection())).append(DELIM);
-
-        String dataStr = new String();
-        // originalId
-
-        for (String oid : data.getOriginalIdList()) {
-            dataStr += cleanDoi(oid) + SEPERATOR; //oid is not openaireid, clean instead of cleanid
-        }
-
-        buff.append(dataStr).append(DELIM);
-        return buff.toString();
-    }
-
-
-    private static String buildDatasource(OafEntity data, String DELIM) {
-        String SEPERATOR = ";";
-
-        StringBuilder buff = new StringBuilder();
-
-        buff.append(getHeader(data, DELIM));
-        Metadata metadata = data.getDatasource().getMetadata();
-
-        //Datasourcetype
-        buff.append(clean(metadata.getDatasourcetype().getClassname())).append(DELIM);
-
-        //Openairecompatibility
-        buff.append(clean(metadata.getOpenairecompatibility().getClassname())).append(DELIM);
-
-        //OfficialName
-        buff.append(clean(metadata.getOfficialname().getValue())).append(DELIM);
-
-        //  Englishname
-        buff.append(clean(metadata.getEnglishname().getValue())).append(DELIM);
-
-        //Websiteurl
-        buff.append(clean(metadata.getWebsiteurl().getValue())).append(DELIM);
-
-        //LogoURL
-        buff.append(clean(metadata.getLogourl().getValue())).append(DELIM);
-
-        //Contactemail
-        buff.append(clean(metadata.getContactemail().getValue())).append(DELIM);
-
-        //Namespaceprefix
-        buff.append(clean(metadata.getNamespaceprefix().getValue())).append(DELIM);
-
-        // latitude
-        buff.append(clean(metadata.getLatitude().getValue())).append(DELIM);
-
-        // longtitude
-        buff.append(clean(metadata.getLongitude().getValue())).append(DELIM);
-
-        // dateofvalidation,
-        buff.append(clean(metadata.getDateofvalidation().getValue())).append(DELIM);
-
-        //Description
-        buff.append(clean(metadata.getDescription().getValue())).append(DELIM);
-
-        //subjects
-        String subj = new String();
-        for (StructuredProperty s : metadata.getSubjectsList()) {
-            subj += clean(s.getValue()) + SEPERATOR;
-        }
-
-        //subjectList
-        buff.append(clean(subj)).append(DELIM);
-
-        //Number of items
-        buff.append(clean(metadata.getOdnumberofitems().getValue())).append(DELIM);
-
-        //Date of number of items
-        buff.append(clean(metadata.getOdnumberofitemsdate().getValue())).append(DELIM);
-
-        // Policies
-        buff.append(clean(metadata.getOdpolicies().getValue())).append(DELIM);
-
-        //languages
-        String dataStr = new String();
-
-        for (StringField lang : metadata.getOdlanguagesList()) {
-            dataStr += clean(lang.getValue()) + SEPERATOR;
-        }
-
-        buff.append(dataStr).append(DELIM);
-        ;
-
-
-        // Content type
-        dataStr = " ";
-        for (StringField c : metadata.getOdcontenttypesList()) {
-            dataStr += clean(c.getValue()) + SEPERATOR;
-        }
-        buff.append(dataStr).append(DELIM);
-
-        //Access info package
-        dataStr = " ";
-
-        for (StringField c : metadata.getAccessinfopackageList()) {
-            dataStr += clean(c.getValue()) + SEPERATOR;
-        }
-
-        buff.append(dataStr).append(DELIM);
-
-        //Release start date
-        buff.append(clean(metadata.getReleasestartdate().getValue())).append(DELIM);
-
-        //Release end date
-        buff.append(clean(metadata.getReleaseenddate().getValue())).append(DELIM);
-
-        //Mission statement url
-        buff.append(clean(metadata.getMissionstatementurl().getValue())).append(DELIM);
-
-        //Data provider
-        buff.append(clean(String.valueOf(metadata.getDataprovider().getValue()))).append(DELIM);
-
-        //Service provider
-        buff.append(clean(String.valueOf(metadata.getServiceprovider().getValue()))).append(DELIM);
-
-        //Database access type
-        buff.append(clean(metadata.getDatabaseaccessrestriction().getValue())).append(DELIM);
-
-        //Data upload type
-        buff.append(clean(metadata.getDatauploadtype().getValue())).append(DELIM);
-
-        //Data upload restrictions
-        buff.append(clean(metadata.getDatauploadrestriction().getValue())).append(DELIM);
-
-        //Versioning
-        buff.append(clean(String.valueOf(metadata.getVersioning().getValue()))).append(DELIM);
-
-        //Citation guideline url
-        buff.append(clean(metadata.getCitationguidelineurl().getValue())).append(DELIM);
-
-        //Quality management kind
-        buff.append(clean(metadata.getQualitymanagementkind().getValue())).append(DELIM);
-
-        //PID systems
-        buff.append(clean(metadata.getPidsystems().getValue())).append(DELIM);
-
-        //Certificates
-        buff.append(clean(metadata.getCertificates().getValue())).append(DELIM);
-
-        //Policies
-        dataStr = " ";
-        for (FieldTypeProtos.KeyValue property : metadata.getPoliciesList()) {
-            dataStr += clean(property.getValue()) + SEPERATOR;
-        }
-
-        buff.append(dataStr).append(DELIM);
-
-        buff.append(getTrust(data)).append(DELIM);
-        return buff.toString();
-    }
-
-
-    private static String buildOrganization(OafEntity data, String DELIM) {
-        String SEPERATOR = ";";
-
-        StringBuilder buff = new StringBuilder();
-        buff.append(getHeader(data, DELIM));
-
-        Organization organization = data.getOrganization();
-        Organization.Metadata metadata = organization.getMetadata();
-
-        //getLegalshortname
-        buff.append(clean(metadata.getLegalshortname().getValue())).append(DELIM);
-        // `name`,
-        buff.append(clean(metadata.getLegalname().getValue())).append(DELIM);
-        //website URL
-        String[] split = metadata.getWebsiteurl().getValue().split(",");
-        String dataStr = new String();
-
-        for (String s : split) {
-            dataStr += s.replace(DELIM, " ") + SEPERATOR;
-        }
-
-        buff.append(dataStr).append(DELIM);
-        //logourl
-        buff.append(clean(metadata.getLogourl().getValue())).append(DELIM);
-        // `country`,
-        buff.append(clean(metadata.getCountry().getClassid())).append(DELIM);
-        buff.append(getTrust(data)).append(DELIM);
-
-        return buff.toString();
-
-    }
-
-    static String getTrust(OafEntity data) {
-        for (FieldTypeProtos.ExtraInfo info : data.getExtraInfoList()) {
-            return (info.getTrust());
-
-        }
-        return " ";
-    }
-
-
-    private static String buildResult(OafEntity data, String DELIM) {
-        String SEPERATOR = ";";
-
-        Result.Metadata metadata = data.getResult().getMetadata();
-
-
-        StringBuilder buff = new StringBuilder();
-
-        buff.append(getHeader(data, DELIM));
-
-        //   titleString
-        String dataStr = new String();
-
-        for (int i = 0; i < metadata.getTitleList().size(); i++) {
-            StructuredProperty title = metadata.getTitleList().get(i);
-            dataStr = clean(title.getValue());
-            break;
-        }
-
-        //  pubtitle
-        buff.append(clean(dataStr)).append(DELIM);
-
-        // date of acceptance CHANGED THIS TO DATE FORMAT
-        buff.append(clean(metadata.getDateofacceptance().getValue())).append(DELIM);
-
-        // publisher
-        buff.append(clean(metadata.getPublisher().getValue())).append(DELIM);
-
-
-        //PID
-        dataStr = " ";
-        for (StructuredProperty p : data.getPidList()) {
-            dataStr += clean(p.getValue()) + SEPERATOR;
-        }
-
-        buff.append(dataStr).append(DELIM);
-
-        //language
-        buff.append(clean(metadata.getLanguage().getClassid())).append(DELIM);
-
-        // RelevantDate
-        dataStr = " ";
-
-        for (StructuredProperty p : metadata.getRelevantdateList()) {
-            dataStr += clean(p.getValue());
-            break;
-        }
-
-        buff.append(dataStr).append(DELIM);
-
-        //Subject
-        dataStr = " ";
-        for (StructuredProperty subj : metadata.getSubjectList()) {
-
-            if (subj.getValue() != null && !subj.getValue().isEmpty()) {
-                dataStr += clean(subj.getValue()) + SEPERATOR;
-            }
-        }
-
-        buff.append(dataStr).append(DELIM);
-
-        //TODO ExternalReference
-
-        buff.append(" ").append(DELIM);
-
-        //Source
-        dataStr = " ";
-        for (StringField s : metadata.getSourceList()) {
-            dataStr += clean(s.getValue()) + SEPERATOR;
-        }
-
-        buff.append(dataStr).append(DELIM);
-
-        //TODO Format     
-        buff.append(" ").append(DELIM);
-
-        //Context
-        dataStr = " ";
-        for (Result.Context s : metadata.getContextList()) {
-            dataStr += clean(s.getId()) + SEPERATOR;
-        }
-        buff.append(dataStr).append(DELIM);
-
-        //country
-
-        String country = " ";
-
-        for (FieldTypeProtos.Qualifier c : metadata.getCountryList()) {
-            country += clean(c.getClassid()) + SEPERATOR;
-        }
-
-        buff.append(country).append(DELIM);
-
-        //Best License
-        buff.append(getBestLicense(data.getResult())).append(DELIM);
-
-        //Description
-        dataStr = " ";
-
-        for (StringField desc : metadata.getDescriptionList()) {
-            dataStr += clean(desc.getValue());
-            break;
-        }
-
-        buff.append(dataStr).append(DELIM);
-
-        //Journal  
-        buff.append(clean(metadata.getJournal().getName())).append(DELIM);  //#null#!
-
-
-        // TODO ERI SOS : HERE IN GET JOUTNAL. GET DATA INFO I CAN FIND PROVENANCE AND SIMILARITY
-
-        // TODO isRelatedTo
-
-        //   resource type
-        buff.append(clean(metadata.getResourcetype().getClassname())).append(DELIM);
-        //   device
-        buff.append(clean(metadata.getDevice().getValue())).append(DELIM);
-        //   size
-        buff.append(clean(metadata.getSize().getValue())).append(DELIM);
-        //     version
-        buff.append(clean(metadata.getVersion().getValue())).append(DELIM);
-
-        //   metadata update
-        buff.append(clean(metadata.getLastmetadataupdate().getValue())).append(DELIM);
-        //   metadata version
-        buff.append(clean(metadata.getMetadataversionnumber().getValue())).append(DELIM);
-
-        // year
-        buff.append(clean(getYearInt(metadata.getDateofacceptance().getValue()))).append(DELIM);
-
-        // type
-        buff.append(clean(metadata.getResulttype().getClassname())).append(DELIM);
-
-        buff.append(getTrust(data)).append(DELIM);
-        return buff.toString();
-    }
-
-
-    private static String buildProject(OafEntity data, String DELIM) {
-        String SEPERATOR = ";";
-
-        StringBuilder buff = new StringBuilder();
-
-        buff.append(getHeader(data, DELIM));
-        Project.Metadata metadata = data.getProject().getMetadata();
-
-
-        //Code
-        buff.append(metadata.getCode().getValue()).append(DELIM);
-        // `Websiteurl`,
-        buff.append(clean(metadata.getWebsiteurl().getValue())).append(DELIM);
-        //TODO here
-
-        // `Acronym`,
-        buff.append(clean(metadata.getAcronym().getValue())).append(DELIM);
-
-        //Title
-        buff.append(clean(metadata.getTitle().getValue())).append(DELIM);
-
-        // Startdate
-        buff.append(clean(metadata.getStartdate().getValue())).append(DELIM);
-
-        // Enddate
-        buff.append(clean(metadata.getEnddate().getValue())).append(DELIM);
-
-        //`Call identifer`
-        buff.append(clean(metadata.getCallidentifier().getValue())).append(DELIM);
-
-        //`KeyWords`
-        buff.append(clean(metadata.getKeywords().getValue())).append(DELIM);
-
-        //`Duration`
-        buff.append(clean(metadata.getDuration().getValue())).append(DELIM);
-
-        //esc39
-        buff.append(clean(metadata.getEcsc39().getValue())).append(DELIM);
-
-        //`Contracttype`
-        buff.append(clean(metadata.getContracttype().getClassname())).append(DELIM);
-
-        //`OA mandate pubs`  TODO DOES NOT EXIST
-        buff.append(clean(metadata.getOamandatepublications().getValue())).append(DELIM);
-        //`Subjects`
-        String dataStr = new String();
-        for (StructuredProperty s : metadata.getSubjectsList()) {
-
-            dataStr += clean(s.getValue()) + SEPERATOR;
-        }
-        buff.append(dataStr).append(DELIM);
-
-        //`EC293`
-        buff.append(clean(metadata.getEcarticle293().getValue())).append(DELIM);
-
-        List<StringField> fundList = metadata.getFundingtreeList();
-
-        if (!fundList.isEmpty()) // `funding_lvl0`,
-        {//TODO funder + 3 funding levels
-           /* funder text,
-            funding_lvl0 text,
-	        funding_lvl1 text,
-	        funding_lvl2 text,
-	        funding_lvl3 text,*/
-            buff.append(FundingParser.getFundingInfo(fundList.get(0).getValue(), DELIM));
-        } else {
-
-            buff.append(FundingParser.getFundingInfo("", DELIM));
-        }
-
-        buff.append(getTrust(data)).append(DELIM);
-
-        return buff.toString();
-
-    }
-
-
-    private static String buildPerson(OafEntity data, String DELIM) {
-        String SEPERATOR = ";";
-
-        PersonProtos.Person person = data.getPerson();
-        PersonProtos.Person.Metadata metadata = person.getMetadata();
-
-        StringBuilder buff = new StringBuilder();
-
-        buff.append(getHeader(data, DELIM));
-
-        // `firstname`,
-        buff.append(clean(metadata.getFirstname().getValue())).append(DELIM);
-
-        // `secondNames`,
-        String dataStr = new String();
-
-        for (StringField s : metadata.getSecondnamesList()) {
-            dataStr += clean(s.getValue()) + ' ';
-        }
-
-        buff.append(dataStr).append(DELIM);
-
-        // `fullname`,
-        buff.append(clean(metadata.getFullname().getValue())).append(DELIM);
-
-        // `Fax`,
-        buff.append(clean(metadata.getFax().getValue())).append(DELIM);
-
-        // `Email`,
-        buff.append(clean(metadata.getEmail().getValue())).append(DELIM);
-
-        // `Phone`,
-        buff.append(clean(metadata.getPhone().getValue())).append(DELIM);
-
-        // `Nationality`,
-        buff.append(clean(metadata.getNationality().getClassid())).append(DELIM);
-
-        // `PIDS`,
-        dataStr = " ";
-        for (StructuredProperty s : data.getPidList()) {
-
-            dataStr += clean(s.getValue()) + ";";
-        }
-        buff.append(dataStr).append(DELIM);
-
-        buff.append(getTrust(data)).append(DELIM);
-
-        return buff.toString();
-
-    }
-
-
-    private static void getResultDatasources(OafEntity valueEntity, String DELIM, Set<String> returnList) {
-        String SEPERATOR = ";";
-
-        Result result = valueEntity.getResult();
-
-        if (valueEntity.getId().contains("dedup")) return;
-
-        //TODO hosted by
-        for (Instance instance : (result.getInstanceList())) {
-            String hostedBy = instance.getHostedby().getKey();
-
-            if (hostedBy != null && !hostedBy.isEmpty()) {
-                returnList.add("resultDatasource" + DELIM + "result" + DELIM +
-                        cleanId(valueEntity.getId()) + DELIM + "datasource" + DELIM + cleanId(hostedBy) + DELIM);
-            }
-        }
-
-        //TODO  collected froms
-        for (FieldTypeProtos.KeyValue collectedFromValue : (valueEntity.getCollectedfromList())) {
-            String collectedFrom = collectedFromValue.getKey();
-            if (collectedFrom != null && !collectedFrom.isEmpty())
-                returnList.add(("resultDatasource" + DELIM
-                        + "result" + DELIM + cleanId(valueEntity.getId()) + DELIM
-                        + "datasource" + DELIM + cleanId(collectedFrom) + DELIM));
-
-        }
-
-
-    }
-
-
-    public static String cleanId(String value) {
-        if (value != null) {
-            //   DO NOT CHANGE THIS: IT REMOVES ID PREFIX ( 5|datacite____::
-            // to datacite____:: )
-            // AND REPLACES OCCURENCES OF DELIM CHARS IN DATA
-            value = value.replaceFirst(".*\\|", "");
-            value = value.replaceAll(ID_PATTERN, "");
-        }
-        return value;
-    }
-
-    private static String clean(String value) {
-        if (value != null) {
-            // TODO DO NOT CHANGE THIS: IT REMOVES ID PREFIX ( 5|datacite____::
-            value = value.replaceAll("[\"\\r\\\\;]", "");
-            value = value.replace("\\", "");
-            value = value.replaceAll(TEXT_PATTERN, " ");
-        }
-
-        return value;
-    }
-
-    private static String cleanDoi(String value) {
-        return value.replaceAll(DOI_PATTERN, "");
-    }
-
-
-    //TODO make them in pairs
-    private static void getDedups(OafEntity valueEntity, String DELIM, Set<String> returnList) {
-        if (!valueEntity.getChildrenList().isEmpty() && valueEntity.getId().contains("dedup")) {
-            ArrayList<String> entries = new ArrayList<String>();
-
-            for (OafEntity child : valueEntity.getChildrenList()) {
-                if (child.getType() == valueEntity.getType() && !child.getId().contains("dedup")) {
-                    // if it is result, then its the deduplication
-                    entries.add(cleanId(child.getId()));
-                }
-            }
-
-            for (int i = 0; i < entries.size() - 1; i++) {
-                for (int j = i + 1; j < entries.size(); j++) {
-                    returnList.add("dedup" + DELIM +
-                            valueEntity.getType().name() + DELIM +
-                            entries.get(i) + DELIM +
-                            valueEntity.getType().name() + DELIM +
-                            entries.get(j) + DELIM);
-                }
-            }
-
-        }
-
-    }
-
-
-    private static String getYearInt(String data) {
-        if (data == null || data.isEmpty() || data.equals("-1")) {
-            return " ";
-        }
-        String[] split = data.split("-");
-
-        if (split == null || split.length == 0) {
-            return " ";
-        }
-
-
-        return split[0];
-
-
-    }
-
-
-    private static String getBestLicense(Result result) {
-        FieldTypeProtos.Qualifier bestLicense = null;
-        LicenseComparator lc = new LicenseComparator();
-        for (Instance instance : (result.getInstanceList())) {
-            if (lc.compare(bestLicense, instance.getLicence()) > 0) {
-                bestLicense = instance.getLicence();
-            }
-        }
-        if (bestLicense != null) {
-            return bestLicense.getClassname();
-        } else {
-            return null;
-        }
-    }
-
-
-}

Property changes on: modules/dnet-openaire-lodexport/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/utils/NewSerializer.java
___________________________________________________________________
Deleted: svn:executable
## -1 +0,0 ##
-*
\ No newline at end of property
Index: modules/dnet-openaire-lodexport/trunk/src/test/java/LodReducerTest.java
===================================================================
--- modules/dnet-openaire-lodexport/trunk/src/test/java/LodReducerTest.java	(nonexistent)
+++ modules/dnet-openaire-lodexport/trunk/src/test/java/LodReducerTest.java	(revision 45699)
@@ -0,0 +1,48 @@
+
+import eu.dnetlib.data.mapreduce.hbase.lodExport.LodReducer;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mrunit.mapreduce.ReduceDriver;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+public class LodReducerTest {
+    ReduceDriver<Text, Text, Text, Text> reduceDriver;
+
+    @Before
+    public void setUp() {
+        Configuration configuration = new Configuration();
+        configuration.set("lod.delim", ",");
+        configuration.set("mapreduce.multipleoutputs.namedOutput.result.key", "org.apache.hadoop.io.Text");
+        configuration.set("mapreduce.multipleoutputs.namedOutput.result.value", "org.apache.hadoop.io.Text");
+        configuration.set("mapreduce.multipleoutputs.namedOutput.result.format", "org.apache.hadoop.mapreduce.lib.output.TextOutputFormat");
+        configuration.set("mapreduce.multipleoutputs", "result");
+
+
+        LodReducer reducer = new LodReducer();
+        //String sourceMappings = "{\"type\":\"result\", \"fields\":[\"<http://purl.org/dc/terms/identifier>\",\"<http://www.eurocris.org/ontologies/cerif/1.3#name>\",\"<http://lod.openaire.eu/vocab/year>\"]}";
+        reduceDriver = ReduceDriver.newReduceDriver(reducer).withConfiguration(configuration);
+
+    }
+
+
+    @Test
+    public void testReducer() throws IOException {
+        ArrayList list = new ArrayList();
+        String value = "doajarticles::9aacfbb6168b5346aa63a49c79a7a88d,2017-01-14T12:40:47.751Z,2016-10-12T12:23:03.305Z,oai:doaj.org/article:85b1bd2dbf344a6cab93f36c7bdfff69;,Zika virus in Brazil and the danger of infestation by Aedes  tegomyia) mosquitoes,2015-01-01, ociedade Brasileira de Medicina Tropical  BMT), 10.1590/0037-8682-0220-2015;,eng, , Aedes aegypti;Aedes albopictus;Arbovirus; tegomyia;Culicidae;Medicine (General);R5-920;Medicine;R;Arctic medicine. Tropical medicine;RC955-962;, , Revista da  ociedade Brasileira de Medicina Tropical, Iss 0 (2015);, , , ,Open Access, Abstract Zika virus, already widely distributed in Africa and Asia, was recently reported in two Northeastern Brazilian:  tate of Bahia and  tate of Rio Grande do Norte, and one  outheastern:  tate of  ão Paulo. This finding adds a potentially noxious virus to a list of several other viruses that are widely transmitted by Aedes  tegomyia) aegypti and Aedes  tegomyia) albopictus in Brazil. The pathology and epidemiology, including the distribution and vectors associated with Zika virus, are reviewed. This review is focused on viruses transmitted by Aedes  tegomyia) mosquitoes, including dengue, Chikungunya, Zika, Mayaro, and yellow fever virus, to emphasize the risks of occurrence for these arboviruses in Brazil and neighboring countries. Other species of Aedes  tegomyia) are discussed, emphasizing their involvement in arbovirus transmission and the possibility of adaptation to environments modified by human activities and introduction in Brazil.,Revista da  ociedade Brasileira de Medicina Tropical,,,,,,,2015,publication, ,";
+        list.add(new Text(value));
+
+        reduceDriver.withInput(new Text("result"), list);
+        reduceDriver.withOutput(new Text("result"), new Text(value));
+        reduceDriver.run();
+    }
+
+
+}
+
+
Index: modules/dnet-openaire-lodexport/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/utils/Serializer.java
===================================================================
--- modules/dnet-openaire-lodexport/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/utils/Serializer.java	(nonexistent)
+++ modules/dnet-openaire-lodexport/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/utils/Serializer.java	(revision 45699)
@@ -0,0 +1,734 @@
+package eu.dnetlib.data.mapreduce.hbase.lodExport.utils;
+
+import eu.dnetlib.data.mapreduce.util.LicenseComparator;
+import eu.dnetlib.data.proto.DatasourceProtos.Datasource.Metadata;
+import eu.dnetlib.data.proto.FieldTypeProtos;
+import eu.dnetlib.data.proto.FieldTypeProtos.StringField;
+import eu.dnetlib.data.proto.FieldTypeProtos.StructuredProperty;
+import eu.dnetlib.data.proto.OafProtos.Oaf;
+import eu.dnetlib.data.proto.OafProtos.OafEntity;
+import eu.dnetlib.data.proto.OafProtos.OafRel;
+import eu.dnetlib.data.proto.OrganizationProtos.Organization;
+import eu.dnetlib.data.proto.PersonProtos;
+import eu.dnetlib.data.proto.ProjectProtos.Project;
+import eu.dnetlib.data.proto.ResultProtos.Result;
+import eu.dnetlib.data.proto.ResultProtos.Result.Instance;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * @author eri Simple serializer that parses input Oaf Protos and prepares them
+ *         for sqoop
+ */
+public class Serializer {
+    private static final String TEXT_PATTERN = "[^a-zA-Z0-9 .-_:/@+=-]";
+    private static final String ID_PATTERN = "[^a-zA-Z0-9.:/_-]";
+    private static final String DOI_PATTERN = ".*\\/(10.)";
+
+    public static String serialize(Oaf oaf, String DELIM) {
+        switch (oaf.getKind()) {
+            case entity:
+                OafEntity valueEntity = oaf.getEntity();
+                switch (valueEntity.getType()) {
+                    case datasource:
+                        return buildDatasource(valueEntity, DELIM);
+                    case organization:
+                        return buildOrganization(valueEntity, DELIM);
+                    case project:
+                        return buildProject(valueEntity, DELIM);
+                    case result:
+                        return buildResult(valueEntity, DELIM);
+                    case person:
+                        return buildPerson(valueEntity, DELIM);
+                    default:
+                        break;
+                }
+                break;
+            case relation:
+                OafRel valueRel = oaf.getRel();
+
+                return serialize(valueRel, DELIM);
+
+        }
+
+        return null;
+
+    }
+
+
+    public static void extractRelations(Oaf oaf, String DELIM, Set<String> relations) {
+        OafEntity valueEntity = oaf.getEntity();
+        switch (valueEntity.getType()) {
+            case result:
+                getResultDatasources(valueEntity, DELIM, relations);
+                getDedups(valueEntity, DELIM, relations);
+            case datasource:
+                getDedups(valueEntity, DELIM, relations);
+            case person:
+                getDedups(valueEntity, DELIM, relations);
+            case organization:
+                getDedups(valueEntity, DELIM, relations);
+            default:
+        }
+
+    }
+
+    public static String serialize(OafRel Rel, String DELIM) {
+        StringBuilder buff;
+        switch (Rel.getRelType()) {
+            case datasourceOrganization:
+                buff = new StringBuilder();
+                buff.append(Rel.getRelType().name()).append(DELIM).append("datasource").append(DELIM).append(cleanId(Rel.getSource())).append(DELIM)
+                        .append("organization").append(DELIM).append(cleanId(Rel.getTarget())).append(DELIM).append("\n");
+                return buff.toString();
+            case resultResult:
+                buff = new StringBuilder();
+                buff.append(Rel.getRelType().name()).append(DELIM).append("result").append(DELIM).append(cleanId(Rel.getSource())).append(DELIM)
+                        .append("result").append(DELIM).append(cleanId(Rel.getTarget())).append(DELIM);
+                return buff.toString();
+            case personPerson:
+                buff = new StringBuilder();
+                buff.append(Rel.getRelType().name()).append(DELIM).append("person").append(DELIM).append(cleanId(Rel.getSource())).append(DELIM)
+                        .append("person").append(DELIM).append(cleanId(Rel.getTarget())).append(DELIM);
+                return buff.toString();
+            case organizationOrganization:
+                buff = new StringBuilder();
+                buff.append(Rel.getRelType().name()).append(DELIM).append("organization").append(DELIM).append(cleanId(Rel.getSource())).append(DELIM)
+                        .append("organization").append(DELIM).append(cleanId(Rel.getTarget())).append(DELIM);
+                return buff.toString();
+            case personResult:
+                buff = new StringBuilder();
+                buff.append(Rel.getRelType().name()).append(DELIM).append("person").append(DELIM).append(cleanId(Rel.getSource())).append(DELIM)
+                        .append("result").append(DELIM).append(cleanId(Rel.getTarget())).append(DELIM);
+                return buff.toString();
+            case projectOrganization:
+                buff = new StringBuilder();
+                buff.append(Rel.getRelType().name()).append(DELIM).append("project").append(DELIM).append(cleanId(Rel.getSource())).append(DELIM)
+                        .append("organization").append(DELIM).append(cleanId(Rel.getTarget())).append(DELIM);
+                return buff.toString();
+            case projectPerson:
+                buff = new StringBuilder();
+                buff.append(Rel.getRelType().name()).append(DELIM).append("project").append(DELIM).append(cleanId(Rel.getSource())).append(DELIM)
+                        .append("person").append(DELIM).append(cleanId(Rel.getTarget())).append(DELIM);
+                return buff.toString();
+            case resultOrganization:
+                buff = new StringBuilder();
+                buff.append(Rel.getRelType().name()).append(DELIM).append("result").append(DELIM).append(cleanId(Rel.getSource())).append(DELIM)
+                        .append("organization").append(DELIM).append(cleanId(Rel.getTarget())).append(DELIM);
+                return buff.toString();
+//TODO - SOS RELATION RESULT PROJECT IS INVERTED! SOURCE IS PROJECT, TARGET IS RESULT
+
+            case resultProject:
+                buff = new StringBuilder();
+                buff.append(Rel.getRelType().name()).append(DELIM).append("result").append(DELIM).append(cleanId(Rel.getTarget())).append(DELIM)
+                        .append("project").append(DELIM).append(cleanId(Rel.getSource())).append(DELIM);
+                return buff.toString();
+            default:
+
+        }
+
+        return "";
+
+    }
+
+    private static String getHeader(OafEntity data, String DELIM) {
+        String SEPERATOR = ";";
+
+        StringBuilder buff = new StringBuilder();
+
+        //  EntityType
+        buff.append(data.getType().name()).append(DELIM);
+
+        // OpenaireID
+        buff.append(cleanId(data.getId())).append(DELIM);
+        //  dateOfTransformation
+
+        // TODO  CRITERIA FOR IDENTIFYING UPDATED RECORDS
+        buff.append(cleanId(data.getDateoftransformation())).append(DELIM);
+
+        //    dateOfCollection
+        buff.append(clean(data.getDateofcollection())).append(DELIM);
+
+        String dataStr = new String();
+        // originalId
+
+        for (String oid : data.getOriginalIdList()) {
+            dataStr += cleanDoi(oid) + SEPERATOR; //oid is not openaireid, clean instead of cleanid
+        }
+
+        buff.append(dataStr).append(DELIM);
+        return buff.toString();
+    }
+
+
+    private static String buildDatasource(OafEntity data, String DELIM) {
+        String SEPERATOR = ";";
+
+        StringBuilder buff = new StringBuilder();
+
+        buff.append(getHeader(data, DELIM));
+        Metadata metadata = data.getDatasource().getMetadata();
+
+        //Datasourcetype
+        buff.append(clean(metadata.getDatasourcetype().getClassname())).append(DELIM);
+
+        //Openairecompatibility
+        buff.append(clean(metadata.getOpenairecompatibility().getClassname())).append(DELIM);
+
+        //OfficialName
+        buff.append(clean(metadata.getOfficialname().getValue())).append(DELIM);
+
+        //  Englishname
+        buff.append(clean(metadata.getEnglishname().getValue())).append(DELIM);
+
+        //Websiteurl
+        buff.append(clean(metadata.getWebsiteurl().getValue())).append(DELIM);
+
+        //LogoURL
+        buff.append(clean(metadata.getLogourl().getValue())).append(DELIM);
+
+        //Contactemail
+        buff.append(clean(metadata.getContactemail().getValue())).append(DELIM);
+
+        //Namespaceprefix
+        buff.append(clean(metadata.getNamespaceprefix().getValue())).append(DELIM);
+
+        // latitude
+        buff.append(clean(metadata.getLatitude().getValue())).append(DELIM);
+
+        // longtitude
+        buff.append(clean(metadata.getLongitude().getValue())).append(DELIM);
+
+        // dateofvalidation,
+        buff.append(clean(metadata.getDateofvalidation().getValue())).append(DELIM);
+
+        //Description
+        buff.append(clean(metadata.getDescription().getValue())).append(DELIM);
+
+        //subjects
+        String subj = new String();
+        for (StructuredProperty s : metadata.getSubjectsList()) {
+            subj += clean(s.getValue()) + SEPERATOR;
+        }
+
+        //subjectList
+        buff.append(clean(subj)).append(DELIM);
+
+        //Number of items
+        buff.append(clean(metadata.getOdnumberofitems().getValue())).append(DELIM);
+
+        //Date of number of items
+        buff.append(clean(metadata.getOdnumberofitemsdate().getValue())).append(DELIM);
+
+        // Policies
+        buff.append(clean(metadata.getOdpolicies().getValue())).append(DELIM);
+
+        //languages
+        String dataStr = new String();
+
+        for (StringField lang : metadata.getOdlanguagesList()) {
+            dataStr += clean(lang.getValue()) + SEPERATOR;
+        }
+
+        buff.append(dataStr).append(DELIM);
+        ;
+
+
+        // Content type
+        dataStr = " ";
+        for (StringField c : metadata.getOdcontenttypesList()) {
+            dataStr += clean(c.getValue()) + SEPERATOR;
+        }
+        buff.append(dataStr).append(DELIM);
+
+        //Access info package
+        dataStr = " ";
+
+        for (StringField c : metadata.getAccessinfopackageList()) {
+            dataStr += clean(c.getValue()) + SEPERATOR;
+        }
+
+        buff.append(dataStr).append(DELIM);
+
+        //Release start date
+        buff.append(clean(metadata.getReleasestartdate().getValue())).append(DELIM);
+
+        //Release end date
+        buff.append(clean(metadata.getReleaseenddate().getValue())).append(DELIM);
+
+        //Mission statement url
+        buff.append(clean(metadata.getMissionstatementurl().getValue())).append(DELIM);
+
+        //Data provider
+        buff.append(clean(String.valueOf(metadata.getDataprovider().getValue()))).append(DELIM);
+
+        //Service provider
+        buff.append(clean(String.valueOf(metadata.getServiceprovider().getValue()))).append(DELIM);
+
+        //Database access type
+        buff.append(clean(metadata.getDatabaseaccessrestriction().getValue())).append(DELIM);
+
+        //Data upload type
+        buff.append(clean(metadata.getDatauploadtype().getValue())).append(DELIM);
+
+        //Data upload restrictions
+        buff.append(clean(metadata.getDatauploadrestriction().getValue())).append(DELIM);
+
+        //Versioning
+        buff.append(clean(String.valueOf(metadata.getVersioning().getValue()))).append(DELIM);
+
+        //Citation guideline url
+        buff.append(clean(metadata.getCitationguidelineurl().getValue())).append(DELIM);
+
+        //Quality management kind
+        buff.append(clean(metadata.getQualitymanagementkind().getValue())).append(DELIM);
+
+        //PID systems
+        buff.append(clean(metadata.getPidsystems().getValue())).append(DELIM);
+
+        //Certificates
+        buff.append(clean(metadata.getCertificates().getValue())).append(DELIM);
+
+        //Policies
+        dataStr = " ";
+        for (FieldTypeProtos.KeyValue property : metadata.getPoliciesList()) {
+            dataStr += clean(property.getValue()) + SEPERATOR;
+        }
+
+        buff.append(dataStr).append(DELIM);
+
+        buff.append(getTrust(data)).append(DELIM);
+        return buff.toString();
+    }
+
+
+    private static String buildOrganization(OafEntity data, String DELIM) {
+        String SEPERATOR = ";";
+
+        StringBuilder buff = new StringBuilder();
+        buff.append(getHeader(data, DELIM));
+
+        Organization organization = data.getOrganization();
+        Organization.Metadata metadata = organization.getMetadata();
+
+        //getLegalshortname
+        buff.append(clean(metadata.getLegalshortname().getValue())).append(DELIM);
+        // `name`,
+        buff.append(clean(metadata.getLegalname().getValue())).append(DELIM);
+        //website URL
+        String[] split = metadata.getWebsiteurl().getValue().split(",");
+        String dataStr = new String();
+
+        for (String s : split) {
+            dataStr += s.replace(DELIM, " ") + SEPERATOR;
+        }
+
+        buff.append(dataStr).append(DELIM);
+        //logourl
+        buff.append(clean(metadata.getLogourl().getValue())).append(DELIM);
+        // `country`,
+        buff.append(clean(metadata.getCountry().getClassid())).append(DELIM);
+        buff.append(getTrust(data)).append(DELIM);
+
+        return buff.toString();
+
+    }
+
+    static String getTrust(OafEntity data) {
+        for (FieldTypeProtos.ExtraInfo info : data.getExtraInfoList()) {
+            return (info.getTrust());
+
+        }
+        return " ";
+    }
+
+
+    private static String buildResult(OafEntity data, String DELIM) {
+        String SEPERATOR = ";";
+
+        Result.Metadata metadata = data.getResult().getMetadata();
+
+
+        StringBuilder buff = new StringBuilder();
+
+        buff.append(getHeader(data, DELIM));
+
+        //   titleString
+        String dataStr = new String();
+
+        for (int i = 0; i < metadata.getTitleList().size(); i++) {
+            StructuredProperty title = metadata.getTitleList().get(i);
+            dataStr = clean(title.getValue());
+            break;
+        }
+
+        //  pubtitle
+        buff.append(clean(dataStr)).append(DELIM);
+
+        // date of acceptance CHANGED THIS TO DATE FORMAT
+        buff.append(clean(metadata.getDateofacceptance().getValue())).append(DELIM);
+
+        // publisher
+        buff.append(clean(metadata.getPublisher().getValue())).append(DELIM);
+
+
+        //PID
+        dataStr = " ";
+        for (StructuredProperty p : data.getPidList()) {
+            dataStr += clean(p.getValue()) + SEPERATOR;
+        }
+
+        buff.append(dataStr).append(DELIM);
+
+        //language
+        buff.append(clean(metadata.getLanguage().getClassid())).append(DELIM);
+
+        // RelevantDate
+        dataStr = " ";
+
+        for (StructuredProperty p : metadata.getRelevantdateList()) {
+            dataStr += clean(p.getValue());
+            break;
+        }
+
+        buff.append(dataStr).append(DELIM);
+
+        //Subject
+        dataStr = " ";
+        for (StructuredProperty subj : metadata.getSubjectList()) {
+
+            if (subj.getValue() != null && !subj.getValue().isEmpty()) {
+                dataStr += clean(subj.getValue()) + SEPERATOR;
+            }
+        }
+
+        buff.append(dataStr).append(DELIM);
+
+        //TODO ExternalReference
+
+        buff.append(" ").append(DELIM);
+
+        //Source
+        dataStr = " ";
+        for (StringField s : metadata.getSourceList()) {
+            dataStr += clean(s.getValue()) + SEPERATOR;
+        }
+
+        buff.append(dataStr).append(DELIM);
+
+        //TODO Format     
+        buff.append(" ").append(DELIM);
+
+        //Context
+        dataStr = " ";
+        for (Result.Context s : metadata.getContextList()) {
+            dataStr += clean(s.getId()) + SEPERATOR;
+        }
+        buff.append(dataStr).append(DELIM);
+
+        //country
+
+        String country = " ";
+
+        for (FieldTypeProtos.Qualifier c : metadata.getCountryList()) {
+            country += clean(c.getClassid()) + SEPERATOR;
+        }
+
+        buff.append(country).append(DELIM);
+
+        //Best License
+        buff.append(getBestLicense(data.getResult())).append(DELIM);
+
+        //Description
+        dataStr = " ";
+
+        for (StringField desc : metadata.getDescriptionList()) {
+            dataStr += clean(desc.getValue());
+            break;
+        }
+
+        buff.append(dataStr).append(DELIM);
+
+        //Journal  
+        buff.append(clean(metadata.getJournal().getName())).append(DELIM);  //#null#!
+
+
+        // TODO ERI SOS : HERE IN GET JOUTNAL. GET DATA INFO I CAN FIND PROVENANCE AND SIMILARITY
+
+        // TODO isRelatedTo
+
+        //   resource type
+        buff.append(clean(metadata.getResourcetype().getClassname())).append(DELIM);
+        //   device
+        buff.append(clean(metadata.getDevice().getValue())).append(DELIM);
+        //   size
+        buff.append(clean(metadata.getSize().getValue())).append(DELIM);
+        //     version
+        buff.append(clean(metadata.getVersion().getValue())).append(DELIM);
+
+        //   metadata update
+        buff.append(clean(metadata.getLastmetadataupdate().getValue())).append(DELIM);
+        //   metadata version
+        buff.append(clean(metadata.getMetadataversionnumber().getValue())).append(DELIM);
+
+        // year
+        buff.append(clean(getYearInt(metadata.getDateofacceptance().getValue()))).append(DELIM);
+
+        // type
+        buff.append(clean(metadata.getResulttype().getClassname())).append(DELIM);
+
+        buff.append(getTrust(data)).append(DELIM);
+        return buff.toString();
+    }
+
+
+    private static String buildProject(OafEntity data, String DELIM) {
+        String SEPERATOR = ";";
+
+        StringBuilder buff = new StringBuilder();
+
+        buff.append(getHeader(data, DELIM));
+        Project.Metadata metadata = data.getProject().getMetadata();
+
+
+        //Code
+        buff.append(metadata.getCode().getValue()).append(DELIM);
+        // `Websiteurl`,
+        buff.append(clean(metadata.getWebsiteurl().getValue())).append(DELIM);
+        //TODO here
+
+        // `Acronym`,
+        buff.append(clean(metadata.getAcronym().getValue())).append(DELIM);
+
+        //Title
+        buff.append(clean(metadata.getTitle().getValue())).append(DELIM);
+
+        // Startdate
+        buff.append(clean(metadata.getStartdate().getValue())).append(DELIM);
+
+        // Enddate
+        buff.append(clean(metadata.getEnddate().getValue())).append(DELIM);
+
+        //`Call identifer`
+        buff.append(clean(metadata.getCallidentifier().getValue())).append(DELIM);
+
+        //`KeyWords`
+        buff.append(clean(metadata.getKeywords().getValue())).append(DELIM);
+
+        //`Duration`
+        buff.append(clean(metadata.getDuration().getValue())).append(DELIM);
+
+        //esc39
+        buff.append(clean(metadata.getEcsc39().getValue())).append(DELIM);
+
+        //`Contracttype`
+        buff.append(clean(metadata.getContracttype().getClassname())).append(DELIM);
+
+        //`OA mandate pubs`  TODO DOES NOT EXIST
+        buff.append(clean(metadata.getOamandatepublications().getValue())).append(DELIM);
+        //`Subjects`
+        String dataStr = new String();
+        for (StructuredProperty s : metadata.getSubjectsList()) {
+
+            dataStr += clean(s.getValue()) + SEPERATOR;
+        }
+        buff.append(dataStr).append(DELIM);
+
+        //`EC293`
+        buff.append(clean(metadata.getEcarticle293().getValue())).append(DELIM);
+
+        List<StringField> fundList = metadata.getFundingtreeList();
+
+        if (!fundList.isEmpty()) // `funding_lvl0`,
+        {//TODO funder + 3 funding levels
+           /* funder text,
+            funding_lvl0 text,
+	        funding_lvl1 text,
+	        funding_lvl2 text,
+	        funding_lvl3 text,*/
+            buff.append(FundingParser.getFundingInfo(fundList.get(0).getValue(), DELIM));
+        } else {
+
+            buff.append(FundingParser.getFundingInfo("", DELIM));
+        }
+
+        buff.append(getTrust(data)).append(DELIM);
+
+        return buff.toString();
+
+    }
+
+
+    private static String buildPerson(OafEntity data, String DELIM) {
+        String SEPERATOR = ";";
+
+        PersonProtos.Person person = data.getPerson();
+        PersonProtos.Person.Metadata metadata = person.getMetadata();
+
+        StringBuilder buff = new StringBuilder();
+
+        buff.append(getHeader(data, DELIM));
+
+        // `firstname`,
+        buff.append(clean(metadata.getFirstname().getValue())).append(DELIM);
+
+        // `secondNames`,
+        String dataStr = new String();
+
+        for (StringField s : metadata.getSecondnamesList()) {
+            dataStr += clean(s.getValue()) + ' ';
+        }
+
+        buff.append(dataStr).append(DELIM);
+
+        // `fullname`,
+        buff.append(clean(metadata.getFullname().getValue())).append(DELIM);
+
+        // `Fax`,
+        buff.append(clean(metadata.getFax().getValue())).append(DELIM);
+
+        // `Email`,
+        buff.append(clean(metadata.getEmail().getValue())).append(DELIM);
+
+        // `Phone`,
+        buff.append(clean(metadata.getPhone().getValue())).append(DELIM);
+
+        // `Nationality`,
+        buff.append(clean(metadata.getNationality().getClassid())).append(DELIM);
+
+        // `PIDS`,
+        dataStr = " ";
+        for (StructuredProperty s : data.getPidList()) {
+
+            dataStr += clean(s.getValue()) + ";";
+        }
+        buff.append(dataStr).append(DELIM);
+
+        buff.append(getTrust(data)).append(DELIM);
+
+        return buff.toString();
+
+    }
+
+
+    private static void getResultDatasources(OafEntity valueEntity, String DELIM, Set<String> returnList) {
+        String SEPERATOR = ";";
+
+        Result result = valueEntity.getResult();
+
+        if (valueEntity.getId().contains("dedup")) return;
+
+        //TODO hosted by
+        for (Instance instance : (result.getInstanceList())) {
+            String hostedBy = instance.getHostedby().getKey();
+
+            if (hostedBy != null && !hostedBy.isEmpty()) {
+                returnList.add("resultDatasource" + DELIM + "result" + DELIM +
+                        cleanId(valueEntity.getId()) + DELIM + "datasource" + DELIM + cleanId(hostedBy) + DELIM);
+            }
+        }
+
+        //TODO  collected froms
+        for (FieldTypeProtos.KeyValue collectedFromValue : (valueEntity.getCollectedfromList())) {
+            String collectedFrom = collectedFromValue.getKey();
+            if (collectedFrom != null && !collectedFrom.isEmpty())
+                returnList.add(("resultDatasource" + DELIM
+                        + "result" + DELIM + cleanId(valueEntity.getId()) + DELIM
+                        + "datasource" + DELIM + cleanId(collectedFrom) + DELIM));
+
+        }
+
+
+    }
+
+
+    public static String cleanId(String value) {
+        if (value != null) {
+            //   DO NOT CHANGE THIS: IT REMOVES ID PREFIX ( 5|datacite____::
+            // to datacite____:: )
+            // AND REPLACES OCCURENCES OF DELIM CHARS IN DATA
+            value = value.replaceFirst(".*\\|", "");
+            value = value.replaceAll(ID_PATTERN, "");
+        }
+        return value;
+    }
+
+    private static String clean(String value) {
+        if (value != null) {
+            // TODO DO NOT CHANGE THIS: IT REMOVES ID PREFIX ( 5|datacite____::
+            value = value.replaceAll("[\"\\r\\\\;]", "");
+            value = value.replace("\\", "");
+            value = value.replaceAll(TEXT_PATTERN, " ");
+        }
+
+        return value;
+    }
+
+    private static String cleanDoi(String value) {
+        return value.replaceAll(DOI_PATTERN, "");
+    }
+
+
+    //TODO make them in pairs
+    private static void getDedups(OafEntity valueEntity, String DELIM, Set<String> returnList) {
+        if (!valueEntity.getChildrenList().isEmpty() && valueEntity.getId().contains("dedup")) {
+            ArrayList<String> entries = new ArrayList<String>();
+
+            for (OafEntity child : valueEntity.getChildrenList()) {
+                if (child.getType() == valueEntity.getType() && !child.getId().contains("dedup")) {
+                    // if it is result, then its the deduplication
+                    entries.add(cleanId(child.getId()));
+                }
+            }
+
+            for (int i = 0; i < entries.size() - 1; i++) {
+                for (int j = i + 1; j < entries.size(); j++) {
+                    returnList.add("dedup" + DELIM +
+                            valueEntity.getType().name() + DELIM +
+                            entries.get(i) + DELIM +
+                            valueEntity.getType().name() + DELIM +
+                            entries.get(j) + DELIM);
+                }
+            }
+
+        }
+
+    }
+
+
+    private static String getYearInt(String data) {
+        if (data == null || data.isEmpty() || data.equals("-1")) {
+            return " ";
+        }
+        String[] split = data.split("-");
+
+        if (split == null || split.length == 0) {
+            return " ";
+        }
+
+
+        return split[0];
+
+
+    }
+
+
+    private static String getBestLicense(Result result) {
+        FieldTypeProtos.Qualifier bestLicense = null;
+        LicenseComparator lc = new LicenseComparator();
+        for (Instance instance : (result.getInstanceList())) {
+            if (lc.compare(bestLicense, instance.getLicence()) > 0) {
+                bestLicense = instance.getLicence();
+            }
+        }
+        if (bestLicense != null) {
+            return bestLicense.getClassname();
+        } else {
+            return null;
+        }
+    }
+
+
+}

Property changes on: modules/dnet-openaire-lodexport/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/utils/Serializer.java
___________________________________________________________________
Added: svn:executable
## -0,0 +1 ##
+*
\ No newline at end of property
Index: modules/dnet-openaire-lodexport/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/LodMapper.java
===================================================================
--- modules/dnet-openaire-lodexport/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/LodMapper.java	(revision 45698)
+++ modules/dnet-openaire-lodexport/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/LodMapper.java	(revision 45699)
@@ -4,7 +4,7 @@
 import eu.dnetlib.data.mapreduce.hbase.index.config.EntityConfigTable;
 import eu.dnetlib.data.mapreduce.hbase.index.config.IndexConfig;
 import eu.dnetlib.data.mapreduce.hbase.index.config.LinkDescriptor;
-import eu.dnetlib.data.mapreduce.hbase.lodExport.utils.NewSerializer;
+import eu.dnetlib.data.mapreduce.hbase.lodExport.utils.Serializer;
 import eu.dnetlib.data.mapreduce.util.OafRowKeyDecoder;
 import eu.dnetlib.data.mapreduce.util.UpdateMerger;
 import eu.dnetlib.data.proto.OafProtos.Oaf;
@@ -30,11 +30,8 @@
  * Mapper Class that reads HBASE contents and prepares them for the StatsDB
  * export
  */
-public class LodMapper extends TableMapper<Text, ImmutableBytesWritable> {
+public class LodMapper extends TableMapper<Text, Text> {
     private Logger log = Logger.getLogger(this.getClass());
-
-    private NewSerializer serializer;
-
     private EntityConfigTable entityConfigTable;
 
     //timeout: ~5 minutes, here in milliseconds
@@ -150,7 +147,7 @@
     }
 
     private void emitEntity(Context context, Oaf oaf, Type type, Oaf.Builder oafBuilder) {
-        String serialized = serializer.Serialize(oafBuilder.build(), DELIM);
+        String serialized = Serializer.serialize(oafBuilder.build(), DELIM);
 
         if (serialized != null && !oaf.getEntity().getId().contains("dedup")) {
 
@@ -157,7 +154,7 @@
             try {
                 //output entity types so we can have seperate files for each type
                 Text TextKeyOut = new Text(oaf.getEntity().getType().toString());
-                context.write((TextKeyOut), new ImmutableBytesWritable(serialized.getBytes()));
+                context.write((TextKeyOut), new Text(serialized));
                 if (!oaf.getEntity().getDateoftransformation().isEmpty()) {
                     context.getCounter(ENTITIES_COUNTER.UPDATED).increment(1);
                 } else {
@@ -189,7 +186,7 @@
 
                         final Map<byte[], byte[]> columnMap = result.getFamilyMap(Bytes.toBytes(ld.getRelDescriptor().getIt()));
 
-                        List<OafRel> relOaf = decodeRelation(oaf, context, columnMap, ld);
+                        List<OafRel> relOaf = decodeRelation(oaf, columnMap, ld);
 
 
                         for (OafRel rel : relOaf) {
@@ -198,10 +195,10 @@
                                 //keep all relations to one file
                                 Text TextKeyOut = new Text("relations");
 
-                                String buff = serializer.Serialize(rel, DELIM);
+                                String buff = Serializer.serialize(rel, DELIM);
 
                                 if (!buff.isEmpty() && !rel.getTarget().contains("dedup")) {
-                                    context.write((TextKeyOut), new ImmutableBytesWritable(buff.getBytes()));
+                                    context.write((TextKeyOut), new Text(buff));
                                     context.getCounter(ENTITIES_COUNTER.TOTAL_RELATIONS).increment(1);
                                 }
 
@@ -212,7 +209,6 @@
                         }
 
                         relOaf.clear();
-                        relOaf = null;
                     } catch (Exception e) {
                         log.error("Error while decoding Relation Proto from HBase Body", e);
 
@@ -225,7 +221,7 @@
 
         Set<String> relationsList = new HashSet<String>();
 
-        serializer.extractRelations(oaf, DELIM, relationsList);
+        Serializer.extractRelations(oaf, DELIM, relationsList);
 
         for (String rel : relationsList) {
             try {
@@ -234,7 +230,7 @@
 
                 if (!oaf.getEntity().getId().contains("dedup")) {
                     if (!rel.contains("dedup")) {
-                        context.write((TextKeyOut), new ImmutableBytesWritable(rel.getBytes()));
+                        context.write((TextKeyOut), new Text(rel));
                         context.getCounter(ENTITIES_COUNTER.TOTAL_RELATIONS).increment(1);
 
                     }
@@ -242,7 +238,7 @@
                     //for dedup entities write only dedup relationships: all the permutations
                     // of children
                     if (rel.contains("dedup")) {
-                        context.write((TextKeyOut), new ImmutableBytesWritable(rel.getBytes()));
+                        context.write((TextKeyOut), new Text(rel));
                         context.getCounter(ENTITIES_COUNTER.TOTAL_RELATIONS).increment(1);
                     }
 
@@ -257,7 +253,7 @@
     }
 
 
-    private ArrayList<OafRel> decodeRelation(final Oaf body, final Context context, Map<byte[], byte[]> columnMap,
+    private ArrayList<OafRel> decodeRelation(final Oaf body, Map<byte[], byte[]> columnMap,
                                              final LinkDescriptor ld) throws IOException, InterruptedException {
 
         ArrayList<OafRel> rels = new ArrayList<OafRel>();
@@ -266,7 +262,7 @@
 
             for (Entry<byte[], byte[]> e : columnMap.entrySet()) {
 
-                final Oaf decodedOaf = decodeProto(context, e.getValue());
+                final Oaf decodedOaf = decodeProto(e.getValue());
                 if (isValid(decodedOaf)) {
                     OafRel.Builder relBuilder = OafRel.newBuilder(decodedOaf.getRel());
 
@@ -286,11 +282,11 @@
         return rels;
     }
 
-    private Oaf decodeProto(final Context context, final byte[] body) {
+    private Oaf decodeProto(final byte[] body) {
         try {
             return Oaf.parseFrom(body);
         } catch (InvalidProtocolBufferException e) {
-
+            log.error("Invalid Protos", e);
         }
         return null;
     }
@@ -322,7 +318,6 @@
 
     @Override
     protected void cleanup(Context context) throws IOException, InterruptedException {
-
         super.cleanup(context);
     }
 
Index: modules/dnet-openaire-lodexport/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/LodReducer.java
===================================================================
--- modules/dnet-openaire-lodexport/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/LodReducer.java	(revision 45698)
+++ modules/dnet-openaire-lodexport/trunk/src/main/java/eu/dnetlib/data/mapreduce/hbase/lodExport/LodReducer.java	(revision 45699)
@@ -8,10 +8,9 @@
 import org.apache.log4j.Logger;
 
 import java.io.IOException;
-import java.nio.charset.Charset;
 import java.util.Iterator;
 
-public class LodReducer extends Reducer<Text, ImmutableBytesWritable, Text, Text> {
+public class LodReducer extends Reducer<Text, Text, Text, Text> {
 
     private Logger log = Logger.getLogger(LodReducer.class);
     private MultipleOutputs MultipleOutputWriter;
@@ -26,18 +25,17 @@
         MultipleOutputWriter = new MultipleOutputs((TaskInputOutputContext) context);
     }
 
-    ;
-
     @Override
-    protected void reduce(final Text key, final Iterable<ImmutableBytesWritable> values, final Context context) throws IOException, InterruptedException {
+    protected void reduce(final Text key, final Iterable<Text> values, final Context context) throws IOException, InterruptedException {
 
-        Iterator<ImmutableBytesWritable> it = values.iterator();
+        Iterator<Text> it = values.iterator();
 
         while (it.hasNext()) {
-            String value = new String(it.next().copyBytes(), Charset.forName("UTF-8"));
+            String value = it.next().toString();
             String[] split = value.split(context.getConfiguration().get("lod.delim"));
             value = value.replaceFirst(split[0] + context.getConfiguration().get("lod.delim"), "");
-            MultipleOutputWriter.write(key.toString(), new Text(split[0].getBytes()), new Text(value.getBytes(Charset.forName("UTF-8"))), key.toString());
+            String basePath = key.toString().equals("relations") ? "rel" : "entities";
+            MultipleOutputWriter.write(key.toString(), new Text(split[0]), new Text(value), basePath);
         }
     }
 
Index: modules/dnet-openaire-lodexport/trunk/pom.xml
===================================================================
--- modules/dnet-openaire-lodexport/trunk/pom.xml	(revision 45698)
+++ modules/dnet-openaire-lodexport/trunk/pom.xml	(revision 45699)
@@ -4,7 +4,7 @@
         <groupId>eu.dnetlib</groupId>
         <artifactId>dnet-parent</artifactId>
         <version>1.0.0</version>
-        <relativePath></relativePath>
+
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <groupId>eu.dnetlib</groupId>
@@ -54,11 +54,11 @@
             <artifactId>slf4j-log4j12</artifactId>
             <version>1.7.6</version>
         </dependency>
-      <dependency>
-			<groupId>com.google.guava</groupId>
-			<artifactId>guava</artifactId>
-			<version>${google.guava.version}</version>
-		</dependency>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <version>${google.guava.version}</version>
+        </dependency>
 
         <dependency>
             <groupId>eu.dnetlib</groupId>
@@ -93,15 +93,21 @@
             <version>${cxf.version}</version>
         </dependency>
         <dependency>
-			<groupId>org.apache.cxf</groupId>
-			<artifactId>cxf-rt-transports-http-jetty</artifactId>
-			<version>${cxf.version}</version>
-		</dependency>
+            <groupId>org.apache.cxf</groupId>
+            <artifactId>cxf-rt-transports-http-jetty</artifactId>
+            <version>${cxf.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.jsoup</groupId>
             <artifactId>jsoup</artifactId>
             <version>1.8.3</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.mrunit</groupId>
+            <artifactId>mrunit</artifactId>
+            <version>0.9.0-incubating</version>
+            <classifier>hadoop2</classifier>
+        </dependency>
     </dependencies>
     <repositories>
         <repository>
