sabeel
/
dnet-hadoop
forked from D-Net/dnet-hadoop
1
0
Fork 0
Code Pull Requests Activity

formatting

This commit is contained in:
Michele Artini 2020-04-29 12:56:58 +02:00
parent a5d7007005
commit c43b4c8962
2 changed files with 473 additions and 508 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

147
dhp-workflows/dhp-graph-mapper/src/main/java/eu/dnetlib/dhp/oa/graph/raw/MigrateDbEntitiesApplication.java
View File

@ -1,3 +1,4 @@
package eu.dnetlib.dhp.oa.graph.raw; package eu.dnetlib.dhp.oa.graph.raw;
import static eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils.asString; import static eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils.asString;
@ -10,6 +11,23 @@ import static eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils.listKeyValues;
import static eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils.qualifier; import static eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils.qualifier;
import static eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils.structuredProperty; import static eu.dnetlib.dhp.oa.graph.raw.common.OafMapperUtils.structuredProperty;
import java.io.Closeable;
import java.io.IOException;
import java.sql.Array;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.function.Consumer;
import java.util.function.Function;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import eu.dnetlib.dhp.application.ArgumentApplicationParser; import eu.dnetlib.dhp.application.ArgumentApplicationParser;
import eu.dnetlib.dhp.oa.graph.raw.common.AbstractMigrationApplication; import eu.dnetlib.dhp.oa.graph.raw.common.AbstractMigrationApplication;
import eu.dnetlib.dhp.oa.graph.raw.common.DbClient; import eu.dnetlib.dhp.oa.graph.raw.common.DbClient;
@ -31,21 +49,6 @@ import eu.dnetlib.dhp.schema.oaf.Relation;
import eu.dnetlib.dhp.schema.oaf.Result; import eu.dnetlib.dhp.schema.oaf.Result;
import eu.dnetlib.dhp.schema.oaf.Software; import eu.dnetlib.dhp.schema.oaf.Software;
import eu.dnetlib.dhp.schema.oaf.StructuredProperty; import eu.dnetlib.dhp.schema.oaf.StructuredProperty;
import java.io.Closeable;
import java.io.IOException;
import java.sql.Array;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.function.Consumer;
import java.util.function.Function;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
public class MigrateDbEntitiesApplication extends AbstractMigrationApplication public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
implements Closeable { implements Closeable {
@ -57,11 +60,10 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
private final long lastUpdateTimestamp; private final long lastUpdateTimestamp;
public static void main(final String[] args) throws Exception { public static void main(final String[] args) throws Exception {
final ArgumentApplicationParser parser = final ArgumentApplicationParser parser = new ArgumentApplicationParser(
new ArgumentApplicationParser( IOUtils
IOUtils.toString( .toString(MigrateDbEntitiesApplication.class
MigrateDbEntitiesApplication.class.getResourceAsStream( .getResourceAsStream("/eu/dnetlib/dhp/oa/graph/migrate_db_entities_parameters.json")));
"/eu/dnetlib/dhp/oa/graph/migrate_db_entities_parameters.json")));
parser.parseArgument(args); parser.parseArgument(args);
@ -71,11 +73,10 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
final String hdfsPath = parser.get("hdfsPath"); final String hdfsPath = parser.get("hdfsPath");
final boolean processClaims = final boolean processClaims = parser.get("action") != null && parser.get("action").equalsIgnoreCase("claims");
parser.get("action") != null && parser.get("action").equalsIgnoreCase("claims");
try (final MigrateDbEntitiesApplication smdbe = try (final MigrateDbEntitiesApplication smdbe = new MigrateDbEntitiesApplication(hdfsPath, dbUrl, dbUser,
new MigrateDbEntitiesApplication(hdfsPath, dbUrl, dbUser, dbPassword)) { dbPassword)) {
if (processClaims) { if (processClaims) {
log.info("Processing claims..."); log.info("Processing claims...");
smdbe.execute("queryClaims.sql", smdbe::processClaims); smdbe.execute("queryClaims.sql", smdbe::processClaims);
@ -115,8 +116,7 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
public void execute(final String sqlFile, final Function<ResultSet, List<Oaf>> producer) public void execute(final String sqlFile, final Function<ResultSet, List<Oaf>> producer)
throws Exception { throws Exception {
final String sql = final String sql = IOUtils.toString(getClass().getResourceAsStream("/eu/dnetlib/dhp/oa/graph/sql/" + sqlFile));
IOUtils.toString(getClass().getResourceAsStream("/eu/dnetlib/dhp/oa/graph/sql/" + sqlFile));
final Consumer<ResultSet> consumer = rs -> producer.apply(rs).forEach(oaf -> emitOaf(oaf)); final Consumer<ResultSet> consumer = rs -> producer.apply(rs).forEach(oaf -> emitOaf(oaf));
@ -133,10 +133,8 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
ds.setId(createOpenaireId(10, rs.getString("datasourceid"), true)); ds.setId(createOpenaireId(10, rs.getString("datasourceid"), true));
ds.setOriginalId(Arrays.asList(rs.getString("datasourceid"))); ds.setOriginalId(Arrays.asList(rs.getString("datasourceid")));
ds.setCollectedfrom( ds
listKeyValues( .setCollectedfrom(listKeyValues(createOpenaireId(10, rs.getString("collectedfromid"), true), rs.getString("collectedfromname")));
createOpenaireId(10, rs.getString("collectedfromid"), true),
rs.getString("collectedfromname")));
ds.setPid(new ArrayList<>()); ds.setPid(new ArrayList<>());
ds.setDateofcollection(asString(rs.getDate("dateofcollection"))); ds.setDateofcollection(asString(rs.getDate("dateofcollection")));
ds.setDateoftransformation(null); // Value not returned by the SQL query ds.setDateoftransformation(null); // Value not returned by the SQL query
@ -176,8 +174,8 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
ds.setPidsystems(field(rs.getString("pidsystems"), info)); ds.setPidsystems(field(rs.getString("pidsystems"), info));
ds.setCertificates(field(rs.getString("certificates"), info)); ds.setCertificates(field(rs.getString("certificates"), info));
ds.setPolicies(new ArrayList<>()); // The sql query returns an empty array ds.setPolicies(new ArrayList<>()); // The sql query returns an empty array
ds.setJournal( ds
prepareJournal(rs.getString("officialname"), rs.getString("journal"), info)); // Journal .setJournal(prepareJournal(rs.getString("officialname"), rs.getString("journal"), info)); // Journal
ds.setDataInfo(info); ds.setDataInfo(info);
ds.setLastupdatetimestamp(lastUpdateTimestamp); ds.setLastupdatetimestamp(lastUpdateTimestamp);
@ -196,10 +194,8 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
p.setId(createOpenaireId(40, rs.getString("projectid"), true)); p.setId(createOpenaireId(40, rs.getString("projectid"), true));
p.setOriginalId(Arrays.asList(rs.getString("projectid"))); p.setOriginalId(Arrays.asList(rs.getString("projectid")));
p.setCollectedfrom( p
listKeyValues( .setCollectedfrom(listKeyValues(createOpenaireId(10, rs.getString("collectedfromid"), true), rs.getString("collectedfromname")));
createOpenaireId(10, rs.getString("collectedfromid"), true),
rs.getString("collectedfromname")));
p.setPid(new ArrayList<>()); p.setPid(new ArrayList<>());
p.setDateofcollection(asString(rs.getDate("dateofcollection"))); p.setDateofcollection(asString(rs.getDate("dateofcollection")));
p.setDateoftransformation(asString(rs.getDate("dateoftransformation"))); p.setDateoftransformation(asString(rs.getDate("dateoftransformation")));
@ -215,8 +211,8 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
p.setKeywords(field(rs.getString("keywords"), info)); p.setKeywords(field(rs.getString("keywords"), info));
p.setDuration(field(Integer.toString(rs.getInt("duration")), info)); p.setDuration(field(Integer.toString(rs.getInt("duration")), info));
p.setEcsc39(field(Boolean.toString(rs.getBoolean("ecsc39")), info)); p.setEcsc39(field(Boolean.toString(rs.getBoolean("ecsc39")), info));
p.setOamandatepublications( p
field(Boolean.toString(rs.getBoolean("oamandatepublications")), info)); .setOamandatepublications(field(Boolean.toString(rs.getBoolean("oamandatepublications")), info));
p.setEcarticle29_3(field(Boolean.toString(rs.getBoolean("ecarticle29_3")), info)); p.setEcarticle29_3(field(Boolean.toString(rs.getBoolean("ecarticle29_3")), info));
p.setSubjects(prepareListOfStructProps(rs.getArray("subjects"), info)); p.setSubjects(prepareListOfStructProps(rs.getArray("subjects"), info));
p.setFundingtree(prepareListFields(rs.getArray("fundingtree"), info)); p.setFundingtree(prepareListFields(rs.getArray("fundingtree"), info));
@ -252,10 +248,8 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
o.setId(createOpenaireId(20, rs.getString("organizationid"), true)); o.setId(createOpenaireId(20, rs.getString("organizationid"), true));
o.setOriginalId(Arrays.asList(rs.getString("organizationid"))); o.setOriginalId(Arrays.asList(rs.getString("organizationid")));
o.setCollectedfrom( o
listKeyValues( .setCollectedfrom(listKeyValues(createOpenaireId(10, rs.getString("collectedfromid"), true), rs.getString("collectedfromname")));
createOpenaireId(10, rs.getString("collectedfromid"), true),
rs.getString("collectedfromname")));
o.setPid(new ArrayList<>()); o.setPid(new ArrayList<>());
o.setDateofcollection(asString(rs.getDate("dateofcollection"))); o.setDateofcollection(asString(rs.getDate("dateofcollection")));
o.setDateoftransformation(asString(rs.getDate("dateoftransformation"))); o.setDateoftransformation(asString(rs.getDate("dateoftransformation")));
@ -269,13 +263,13 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
o.setEclegalbody(field(Boolean.toString(rs.getBoolean("eclegalbody")), info)); o.setEclegalbody(field(Boolean.toString(rs.getBoolean("eclegalbody")), info));
o.setEclegalperson(field(Boolean.toString(rs.getBoolean("eclegalperson")), info)); o.setEclegalperson(field(Boolean.toString(rs.getBoolean("eclegalperson")), info));
o.setEcnonprofit(field(Boolean.toString(rs.getBoolean("ecnonprofit")), info)); o.setEcnonprofit(field(Boolean.toString(rs.getBoolean("ecnonprofit")), info));
o.setEcresearchorganization( o
field(Boolean.toString(rs.getBoolean("ecresearchorganization")), info)); .setEcresearchorganization(field(Boolean.toString(rs.getBoolean("ecresearchorganization")), info));
o.setEchighereducation(field(Boolean.toString(rs.getBoolean("echighereducation")), info)); o.setEchighereducation(field(Boolean.toString(rs.getBoolean("echighereducation")), info));
o.setEcinternationalorganizationeurinterests( o
field(Boolean.toString(rs.getBoolean("ecinternationalorganizationeurinterests")), info)); .setEcinternationalorganizationeurinterests(field(Boolean.toString(rs.getBoolean("ecinternationalorganizationeurinterests")), info));
o.setEcinternationalorganization( o
field(Boolean.toString(rs.getBoolean("ecinternationalorganization")), info)); .setEcinternationalorganization(field(Boolean.toString(rs.getBoolean("ecinternationalorganization")), info));
o.setEcenterprise(field(Boolean.toString(rs.getBoolean("ecenterprise")), info)); o.setEcenterprise(field(Boolean.toString(rs.getBoolean("ecenterprise")), info));
o.setEcsmevalidated(field(Boolean.toString(rs.getBoolean("ecsmevalidated")), info)); o.setEcsmevalidated(field(Boolean.toString(rs.getBoolean("ecsmevalidated")), info));
o.setEcnutscode(field(Boolean.toString(rs.getBoolean("ecnutscode")), info)); o.setEcnutscode(field(Boolean.toString(rs.getBoolean("ecnutscode")), info));
@ -294,10 +288,7 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
final DataInfo info = prepareDataInfo(rs); final DataInfo info = prepareDataInfo(rs);
final String orgId = createOpenaireId(20, rs.getString("organization"), true); final String orgId = createOpenaireId(20, rs.getString("organization"), true);
final String dsId = createOpenaireId(10, rs.getString("datasource"), true); final String dsId = createOpenaireId(10, rs.getString("datasource"), true);
final List<KeyValue> collectedFrom = final List<KeyValue> collectedFrom = listKeyValues(createOpenaireId(10, rs.getString("collectedfromid"), true), rs.getString("collectedfromname"));
listKeyValues(
createOpenaireId(10, rs.getString("collectedfromid"), true),
rs.getString("collectedfromname"));
final Relation r1 = new Relation(); final Relation r1 = new Relation();
r1.setRelType("datasourceOrganization"); r1.setRelType("datasourceOrganization");
@ -330,10 +321,7 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
final DataInfo info = prepareDataInfo(rs); final DataInfo info = prepareDataInfo(rs);
final String orgId = createOpenaireId(20, rs.getString("resporganization"), true); final String orgId = createOpenaireId(20, rs.getString("resporganization"), true);
final String projectId = createOpenaireId(40, rs.getString("project"), true); final String projectId = createOpenaireId(40, rs.getString("project"), true);
final List<KeyValue> collectedFrom = final List<KeyValue> collectedFrom = listKeyValues(createOpenaireId(10, rs.getString("collectedfromid"), true), rs.getString("collectedfromname"));
listKeyValues(
createOpenaireId(10, rs.getString("collectedfromid"), true),
rs.getString("collectedfromname"));
final Relation r1 = new Relation(); final Relation r1 = new Relation();
r1.setRelType("projectOrganization"); r1.setRelType("projectOrganization");
@ -364,17 +352,9 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
public List<Oaf> processClaims(final ResultSet rs) { public List<Oaf> processClaims(final ResultSet rs) {
final DataInfo info = final DataInfo info =
dataInfo( dataInfo(false, null, false, false, qualifier("user:claim", "user:claim", "dnet:provenanceActions", "dnet:provenanceActions"), "0.9");
false,
null,
false,
false,
qualifier(
"user:claim", "user:claim", "dnet:provenanceActions", "dnet:provenanceActions"),
"0.9");
final List<KeyValue> collectedFrom = final List<KeyValue> collectedFrom = listKeyValues(createOpenaireId(10, "infrastruct_::openaire", true), "OpenAIRE");
listKeyValues(createOpenaireId(10, "infrastruct_::openaire", true), "OpenAIRE");
try { try {
@ -402,10 +382,8 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
return Arrays.asList(r); return Arrays.asList(r);
} else { } else {
final String sourceId = final String sourceId = createOpenaireId(rs.getString("source_type"), rs.getString("source_id"), false);
createOpenaireId(rs.getString("source_type"), rs.getString("source_id"), false); final String targetId = createOpenaireId(rs.getString("target_type"), rs.getString("target_id"), false);
final String targetId =
createOpenaireId(rs.getString("target_type"), rs.getString("target_id"), false);
final Relation r1 = new Relation(); final Relation r1 = new Relation();
final Relation r2 = new Relation(); final Relation r2 = new Relation();
@ -462,19 +440,11 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
final String inferenceprovenance = rs.getString("inferenceprovenance"); final String inferenceprovenance = rs.getString("inferenceprovenance");
final Boolean inferred = rs.getBoolean("inferred"); final Boolean inferred = rs.getBoolean("inferred");
final String trust = rs.getString("trust"); final String trust = rs.getString("trust");
return dataInfo( return dataInfo(deletedbyinference, inferenceprovenance, inferred, false, MigrationConstants.ENTITYREGISTRY_PROVENANCE_ACTION, trust);
deletedbyinference,
inferenceprovenance,
inferred,
false,
MigrationConstants.ENTITYREGISTRY_PROVENANCE_ACTION,
trust);
} }
private Qualifier prepareQualifierSplitting(final String s) { private Qualifier prepareQualifierSplitting(final String s) {
if (StringUtils.isBlank(s)) { if (StringUtils.isBlank(s)) { return null; }
return null;
}
final String[] arr = s.split("@@@"); final String[] arr = s.split("@@@");
return arr.length == 4 ? qualifier(arr[0], arr[1], arr[2], arr[3]) : null; return arr.length == 4 ? qualifier(arr[0], arr[1], arr[2], arr[3]) : null;
} }
@ -488,22 +458,19 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
} }
private StructuredProperty prepareStructProp(final String s, final DataInfo dataInfo) { private StructuredProperty prepareStructProp(final String s, final DataInfo dataInfo) {
if (StringUtils.isBlank(s)) { if (StringUtils.isBlank(s)) { return null; }
return null;
}
final String[] parts = s.split("###"); final String[] parts = s.split("###");
if (parts.length == 2) { if (parts.length == 2) {
final String value = parts[0]; final String value = parts[0];
final String[] arr = parts[1].split("@@@"); final String[] arr = parts[1].split("@@@");
if (arr.length == 4) { if (arr.length == 4) { return structuredProperty(value, arr[0], arr[1], arr[2], arr[3], dataInfo); }
return structuredProperty(value, arr[0], arr[1], arr[2], arr[3], dataInfo);
}
} }
return null; return null;
} }
private List<StructuredProperty> prepareListOfStructProps( private List<StructuredProperty> prepareListOfStructProps(
final Array array, final DataInfo dataInfo) throws SQLException { final Array array,
final DataInfo dataInfo) throws SQLException {
final List<StructuredProperty> res = new ArrayList<>(); final List<StructuredProperty> res = new ArrayList<>();
if (array != null) { if (array != null) {
for (final String s : (String[]) array.getArray()) { for (final String s : (String[]) array.getArray()) {
@ -522,10 +489,8 @@ public class MigrateDbEntitiesApplication extends AbstractMigrationApplication
final String[] arr = sj.split("@@@"); final String[] arr = sj.split("@@@");
if (arr.length == 3) { if (arr.length == 3) {
final String issn = StringUtils.isNotBlank(arr[0]) ? arr[0] : null; final String issn = StringUtils.isNotBlank(arr[0]) ? arr[0] : null;
final String eissn = StringUtils.isNotBlank(arr[1]) ? arr[1] : null; final String eissn = StringUtils.isNotBlank(arr[1]) ? arr[1] : null;;
; final String lissn = StringUtils.isNotBlank(arr[2]) ? arr[2] : null;;
final String lissn = StringUtils.isNotBlank(arr[2]) ? arr[2] : null;
;
if (issn != null || eissn != null || lissn != null) { if (issn != null || eissn != null || lissn != null) {
return journal(name, issn, eissn, eissn, null, null, null, null, null, null, null, info); return journal(name, issn, eissn, eissn, null, null, null, null, null, null, null, info);
} }

4
dhp-workflows/dhp-stats-update/pom.xml
View File

@ -3,10 +3,10 @@
<parent> <parent>
<artifactId>dhp-workflows</artifactId> <artifactId>dhp-workflows</artifactId>
<groupId>eu.dnetlib.dhp</groupId> <groupId>eu.dnetlib.dhp</groupId>
<version>1.1.6-SNAPSHOT</version> <version>1.1.7-SNAPSHOT</version>
</parent> </parent>
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>
<artifactId>dhp-statistics-workflow</artifactId> <artifactId>dhp-stats-update</artifactId>
<dependencies> <dependencies>
<dependency> <dependency>
<groupId>org.apache.spark</groupId> <groupId>org.apache.spark</groupId>