spyros
/
dnet-hadoop
forked from D-Net/dnet-hadoop
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

different date format in lambda file parsing

This commit is contained in:
Enrico Ottonello 2020-05-11 14:41:11 +02:00
parent b90609848b
commit 7990894454
1 changed files with 164 additions and 143 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

307
dhp-workflows/dhp-doiboost/src/main/java/eu/dnetlib/doiboost/orcid/OrcidDownloader.java
View File

@ -1,14 +1,14 @@
package eu.dnetlib.doiboost.orcid; package eu.dnetlib.doiboost.orcid;
import eu.dnetlib.dhp.application.ArgumentApplicationParser;
import java.io.BufferedReader; import java.io.BufferedReader;
import java.io.IOException; import java.io.IOException;
import java.io.InputStreamReader; import java.io.InputStreamReader;
import java.text.ParseException;
import java.text.SimpleDateFormat; import java.text.SimpleDateFormat;
import java.util.Arrays; import java.util.Arrays;
import java.util.Date; import java.util.Date;
import java.util.List; import java.util.List;
import org.apache.commons.io.IOUtils; import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FSDataInputStream;
@ -22,158 +22,179 @@ import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients; import org.apache.http.impl.client.HttpClients;
import org.mortbay.log.Log; import org.mortbay.log.Log;
import eu.dnetlib.dhp.application.ArgumentApplicationParser;
public class OrcidDownloader extends OrcidDSManager { public class OrcidDownloader extends OrcidDSManager {
static final int REQ_LIMIT = 24; static final int REQ_LIMIT = 24;
static final int REQ_MAX_TEST = 100; static final int REQ_MAX_TEST = 100;
static final int RECORD_PARSED_COUNTER_LOG_INTERVAL = 50000; static final int RECORD_PARSED_COUNTER_LOG_INTERVAL = 10;
static final String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss."; static final String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";
static final String lastUpdate = "2019-09-30 00:00:00.000000"; static final String lastUpdate = "2019-09-30 00:00:00";
private String lambdaFileName; private String lambdaFileName;
private String outputPath; private String outputPath;
private String token; private String token;
public static void main(String[] args) throws IOException, Exception { public static void main(String[] args) throws IOException, Exception {
OrcidDownloader orcidDownloader = new OrcidDownloader(); OrcidDownloader orcidDownloader = new OrcidDownloader();
orcidDownloader.loadArgs(args); orcidDownloader.loadArgs(args);
orcidDownloader.parseLambdaFile(); orcidDownloader.parseLambdaFile();
} }
private String downloadRecord(String orcidId) throws Exception { private String downloadRecord(String orcidId) {
try (CloseableHttpClient client = HttpClients.createDefault()) { try (CloseableHttpClient client = HttpClients.createDefault()) {
HttpGet httpGet = new HttpGet("https://api.orcid.org/v3.0/" + orcidId + "/record"); HttpGet httpGet = new HttpGet("https://api.orcid.org/v3.0/" + orcidId + "/record");
httpGet.addHeader("Accept", "application/vnd.orcid+xml"); httpGet.addHeader("Accept", "application/vnd.orcid+xml");
httpGet.addHeader("Authorization", String.format("Bearer %s", token)); httpGet.addHeader("Authorization", String.format("Bearer %s", token));
CloseableHttpResponse response = client.execute(httpGet); CloseableHttpResponse response = client.execute(httpGet);
if (response.getStatusLine().getStatusCode() != 200) { if (response.getStatusLine().getStatusCode() != 200) {
Log.warn( Log
"Downloading " + orcidId + " status code: " + response.getStatusLine().getStatusCode()); .warn(
return new String(""); "Downloading " + orcidId + " status code: " + response.getStatusLine().getStatusCode());
} return new String("");
return IOUtils.toString(response.getEntity().getContent()); }
return IOUtils.toString(response.getEntity().getContent());
} catch (Throwable e) { } catch (Throwable e) {
Log.warn("Downloading " + orcidId, e); Log.warn("Downloading " + orcidId, e.getMessage());
} }
return new String(""); return new String("");
} }
public void parseLambdaFile() throws Exception { public void parseLambdaFile() throws Exception {
int parsedRecordsCounter = 0; int parsedRecordsCounter = 0;
int downloadedRecordsCounter = 0; int downloadedRecordsCounter = 0;
int savedRecordsCounter = 0; int savedRecordsCounter = 0;
long startDownload = 0; long startDownload = 0;
Configuration conf = initConfigurationObject(); Configuration conf = initConfigurationObject();
FileSystem fs = initFileSystemObject(conf); FileSystem fs = initFileSystemObject(conf);
String lambdaFileUri = hdfsServerUri.concat(hdfsOrcidDefaultPath).concat(lambdaFileName); String lambdaFileUri = hdfsServerUri.concat(hdfsOrcidDefaultPath).concat(lambdaFileName);
Path hdfsreadpath = new Path(lambdaFileUri); Path hdfsreadpath = new Path(lambdaFileUri);
FSDataInputStream lambdaFileStream = fs.open(hdfsreadpath); FSDataInputStream lambdaFileStream = fs.open(hdfsreadpath);
Path hdfsoutputPath = Path hdfsoutputPath = new Path(
new Path( hdfsServerUri
hdfsServerUri .concat(hdfsOrcidDefaultPath)
.concat(hdfsOrcidDefaultPath) .concat(outputPath)
.concat(outputPath) .concat("orcid_records.seq"));
.concat("orcid_records.seq"));
try (SequenceFile.Writer writer = try (SequenceFile.Writer writer = SequenceFile
SequenceFile.createWriter( .createWriter(
conf, conf,
SequenceFile.Writer.file(hdfsoutputPath), SequenceFile.Writer.file(hdfsoutputPath),
SequenceFile.Writer.keyClass(Text.class), SequenceFile.Writer.keyClass(Text.class),
SequenceFile.Writer.valueClass(Text.class))) { SequenceFile.Writer.valueClass(Text.class))) {
try (BufferedReader br = new BufferedReader(new InputStreamReader(lambdaFileStream))) { try (BufferedReader br = new BufferedReader(new InputStreamReader(lambdaFileStream))) {
String line; String line;
int nReqTmp = 0; int nReqTmp = 0;
startDownload = System.currentTimeMillis(); startDownload = System.currentTimeMillis();
long startReqTmp = System.currentTimeMillis(); long startReqTmp = System.currentTimeMillis();
while ((line = br.readLine()) != null) { while ((line = br.readLine()) != null) {
parsedRecordsCounter++; parsedRecordsCounter++;
// skip headers line // skip headers line
if (parsedRecordsCounter == 1) { if (parsedRecordsCounter == 1) {
continue; continue;
} }
String[] values = line.split(","); String[] values = line.split(",");
List<String> recordInfo = Arrays.asList(values); List<String> recordInfo = Arrays.asList(values);
if (isModified(recordInfo.get(3))) { String orcidId = recordInfo.get(0);
String record = downloadRecord(recordInfo.get(0)); if (isModified(orcidId, recordInfo.get(3))) {
downloadedRecordsCounter++; String record = downloadRecord(orcidId);
if (!record.isEmpty()) { downloadedRecordsCounter++;
String compressRecord = ArgumentApplicationParser.compressArgument(record); if (!record.isEmpty()) {
final Text key = new Text(recordInfo.get(0)); String compressRecord = ArgumentApplicationParser.compressArgument(record);
final Text value = new Text(compressRecord); final Text key = new Text(recordInfo.get(0));
final Text value = new Text(compressRecord);
try { try {
writer.append(key, value); writer.append(key, value);
savedRecordsCounter++; savedRecordsCounter++;
} catch (IOException e) { } catch (IOException e) {
Log.debug("Writing to sequence file: " + e.getMessage()); Log.warn("Writing to sequence file: " + e.getMessage());
Log.debug(e); Log.warn(e);
throw new RuntimeException(e); throw new RuntimeException(e);
} }
} }
} }
long endReq = System.currentTimeMillis(); long endReq = System.currentTimeMillis();
nReqTmp++; nReqTmp++;
if (nReqTmp == REQ_LIMIT) { if (nReqTmp == REQ_LIMIT) {
long reqSessionDuration = endReq - startReqTmp; long reqSessionDuration = endReq - startReqTmp;
if (reqSessionDuration <= 1000) { if (reqSessionDuration <= 1000) {
Log.warn( Log
"\nreqSessionDuration: " .warn(
+ reqSessionDuration "\nreqSessionDuration: "
+ " nReqTmp: " + reqSessionDuration
+ nReqTmp + " nReqTmp: "
+ " wait ...."); + nReqTmp
Thread.sleep(1000 - reqSessionDuration); + " wait ....");
} else { Thread.sleep(1000 - reqSessionDuration);
nReqTmp = 0; } else {
startReqTmp = System.currentTimeMillis(); nReqTmp = 0;
} startReqTmp = System.currentTimeMillis();
} }
}
// if (parsedRecordsCounter>REQ_MAX_TEST) { if (parsedRecordsCounter > REQ_MAX_TEST) {
// break; break;
// } }
if ((parsedRecordsCounter % RECORD_PARSED_COUNTER_LOG_INTERVAL) == 0) { if ((parsedRecordsCounter % RECORD_PARSED_COUNTER_LOG_INTERVAL) == 0) {
Log.info("Current record parsed: " + parsedRecordsCounter); Log
Log.info("Current record downloaded: " + downloadedRecordsCounter); .info(
Log.info("Current record saved: " + savedRecordsCounter); "Current parsed: "
} + parsedRecordsCounter
} + " downloaded: "
long endDownload = System.currentTimeMillis(); + downloadedRecordsCounter
long downloadTime = endDownload - startDownload; + " saved: "
Log.info("Download time: " + ((downloadTime / 1000) / 60) + " minutes"); + savedRecordsCounter);
} }
} }
lambdaFileStream.close(); long endDownload = System.currentTimeMillis();
Log.info("Download started at: " + new Date(startDownload).toString()); long downloadTime = endDownload - startDownload;
Log.info("Parsed Records Counter: " + parsedRecordsCounter); Log.info("Download time: " + ((downloadTime / 1000) / 60) + " minutes");
Log.info("Downloaded Records Counter: " + downloadedRecordsCounter); }
Log.info("Saved Records Counter: " + savedRecordsCounter); }
} lambdaFileStream.close();
Log.info("Download started at: " + new Date(startDownload).toString());
Log.info("Parsed Records Counter: " + parsedRecordsCounter);
Log.info("Downloaded Records Counter: " + downloadedRecordsCounter);
Log.info("Saved Records Counter: " + savedRecordsCounter);
}
private void loadArgs(String[] args) throws IOException, Exception { private void loadArgs(String[] args) throws IOException, Exception {
final ArgumentApplicationParser parser = final ArgumentApplicationParser parser = new ArgumentApplicationParser(
new ArgumentApplicationParser( IOUtils
IOUtils.toString( .toString(
OrcidDownloader.class.getResourceAsStream( OrcidDownloader.class
"/eu/dnetlib/dhp/doiboost/download_orcid_data.json"))); .getResourceAsStream(
parser.parseArgument(args); "/eu/dnetlib/dhp/doiboost/download_orcid_data.json")));
parser.parseArgument(args);
hdfsServerUri = parser.get("hdfsServerUri"); hdfsServerUri = parser.get("hdfsServerUri");
Log.info("HDFS URI: " + hdfsServerUri); Log.info("HDFS URI: " + hdfsServerUri);
hdfsOrcidDefaultPath = parser.get("hdfsOrcidDefaultPath"); hdfsOrcidDefaultPath = parser.get("hdfsOrcidDefaultPath");
Log.info("Default Path: " + hdfsOrcidDefaultPath); Log.info("Default Path: " + hdfsOrcidDefaultPath);
lambdaFileName = parser.get("lambdaFileName"); lambdaFileName = parser.get("lambdaFileName");
Log.info("Lambda File Name: " + lambdaFileName); Log.info("Lambda File Name: " + lambdaFileName);
outputPath = parser.get("outputPath"); outputPath = parser.get("outputPath");
Log.info("Output Data: " + outputPath); Log.info("Output Data: " + outputPath);
token = parser.get("token"); token = parser.get("token");
} }
private boolean isModified(String modifiedDate) throws ParseException { private boolean isModified(String orcidId, String modifiedDate) {
Date modifiedDateDt = new SimpleDateFormat(DATE_FORMAT).parse(modifiedDate); Date modifiedDateDt = null;
Date lastUpdateDt = new SimpleDateFormat(DATE_FORMAT).parse(lastUpdate); Date lastUpdateDt = null;
return modifiedDateDt.after(lastUpdateDt); try {
} if (modifiedDate.length() != 19) {
modifiedDate = modifiedDate.substring(0, 19);
}
modifiedDateDt = new SimpleDateFormat(DATE_FORMAT).parse(modifiedDate);
lastUpdateDt = new SimpleDateFormat(DATE_FORMAT).parse(lastUpdate);
} catch (Exception e) {
Log.warn("[" + orcidId + "] Parsing date: ", e.getMessage());
return true;
}
return modifiedDateDt.after(lastUpdateDt);
}
} }