mdstore read lock/unlock when bulk copying records from mongodb to hdfs

This commit is contained in:
Claudio Atzori 2021-05-04 18:06:21 +02:00
parent ba86835951
commit 923d19ea8e
10 changed files with 276 additions and 51 deletions

View File

@ -3,26 +3,19 @@ package eu.dnetlib.dhp.common;
import java.io.Closeable; import java.io.Closeable;
import java.io.IOException; import java.io.IOException;
import java.util.ArrayList; import java.util.*;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.stream.StreamSupport; import java.util.stream.StreamSupport;
import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.bson.Document; import org.bson.Document;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import com.google.common.collect.Iterables; import com.google.common.collect.Iterables;
import com.mongodb.BasicDBObject; import com.mongodb.*;
import com.mongodb.MongoClient;
import com.mongodb.MongoClientURI;
import com.mongodb.QueryBuilder;
import com.mongodb.client.MongoCollection; import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase; import com.mongodb.client.MongoDatabase;
import com.mongodb.client.MongoIterable;
public class MdstoreClient implements Closeable { public class MdstoreClient implements Closeable {
@ -31,37 +24,96 @@ public class MdstoreClient implements Closeable {
private final MongoClient client; private final MongoClient client;
private final MongoDatabase db; private final MongoDatabase db;
public static final String MD_ID = "mdId";
public static final String CURRENT_ID = "currentId";
public static final String EXPIRING = "expiring";
public static final String ID = "id";
public static final String LAST_READ = "lastRead";
public static final String FORMAT = "format";
public static final String LAYOUT = "layout";
public static final String INTERPRETATION = "interpretation";
public static final String BODY = "body";
private static final String COLL_METADATA = "metadata"; private static final String COLL_METADATA = "metadata";
private static final String COLL_METADATA_MANAGER = "metadataManager"; private static final String COLL_METADATA_MANAGER = "metadataManager";
public MdstoreClient(final String baseUrl, final String dbName) { public MdstoreClient(final MongoClient mongoClient, final String dbName) {
this.client = new MongoClient(new MongoClientURI(baseUrl)); this.client = mongoClient;
this.db = getDb(client, dbName); this.db = getDb(client, dbName);
} }
public Iterable<String> mdStoreRecords(final String mdId) {
return recordIterator(mdStore(mdId));
}
public MongoCollection<Document> mdStore(final String mdId) { public MongoCollection<Document> mdStore(final String mdId) {
BasicDBObject query = (BasicDBObject) QueryBuilder.start("mdId").is(mdId).get(); final Document mdStoreInfo = getMDStoreInfo(mdId);
final String currentId = mdStoreInfo.getString(CURRENT_ID);
log.info("querying current mdId: {}", query.toJson()); log.info("reading currentId: {}", currentId);
final String currentId = Optional
.ofNullable(getColl(db, COLL_METADATA_MANAGER, true).find(query))
.map(r -> r.first())
.map(d -> d.getString("currentId"))
.orElseThrow(() -> new IllegalArgumentException("cannot find current mdstore id for: " + mdId));
log.info("currentId: {}", currentId);
return getColl(db, currentId, true); return getColl(db, currentId, true);
} }
public MdstoreTx readLock(final String mdId) {
final Document mdStoreInfo = getMDStoreInfo(mdId);
final List expiring = mdStoreInfo.get(EXPIRING, List.class);
final String currentId = mdStoreInfo.getString(CURRENT_ID);
log.info("locking collection {}", currentId);
if (expiring.size() > 0) {
for (Object value : expiring) {
final Document obj = (Document) value;
final String expiringId = (String) obj.get(ID);
if (currentId.equals(expiringId)) {
obj.put(LAST_READ, new Date());
break;
}
}
} else {
final BasicDBObject readStore = new BasicDBObject();
readStore.put(ID, currentId);
readStore.put(LAST_READ, new Date());
expiring.add(readStore);
}
getColl(db, COLL_METADATA_MANAGER, true)
.findOneAndReplace(new BasicDBObject("_id", mdStoreInfo.get("_id")), mdStoreInfo);
return new MdstoreTx(this, mdId, currentId);
}
public void readUnlock(final String mdId, final String currentId) {
log.info("unlocking collection {}", currentId);
final Document mdStoreInfo = getMDStoreInfo(mdId);
final List<Document> expiring = mdStoreInfo.get(EXPIRING, List.class);
expiring
.stream()
.filter(d -> currentId.equals(d.getString(ID)))
.findFirst()
.ifPresent(expired -> expiring.remove(expired));
}
/**
* Retrieves from the MDStore mongoDB database a snapshot of the [mdID, currentID] pairs.
* @param mdFormat
* @param mdLayout
* @param mdInterpretation
* @return an HashMap of the mdID -> currentID associations.
*/
public Map<String, String> validCollections( public Map<String, String> validCollections(
final String mdFormat, final String mdLayout, final String mdInterpretation) { final String mdFormat, final String mdLayout, final String mdInterpretation) {
final Map<String, String> transactions = new HashMap<>(); final Map<String, String> transactions = new HashMap<>();
for (final Document entry : getColl(db, COLL_METADATA_MANAGER, true).find()) { for (final Document entry : getColl(db, COLL_METADATA_MANAGER, true).find()) {
final String mdId = entry.getString("mdId"); final String mdId = entry.getString(MD_ID);
final String currentId = entry.getString("currentId"); final String currentId = entry.getString(CURRENT_ID);
if (StringUtils.isNoneBlank(mdId, currentId)) { if (StringUtils.isNoneBlank(mdId, currentId)) {
transactions.put(mdId, currentId); transactions.put(mdId, currentId);
} }
@ -69,11 +121,11 @@ public class MdstoreClient implements Closeable {
final Map<String, String> res = new HashMap<>(); final Map<String, String> res = new HashMap<>();
for (final Document entry : getColl(db, COLL_METADATA, true).find()) { for (final Document entry : getColl(db, COLL_METADATA, true).find()) {
if (entry.getString("format").equals(mdFormat) if (entry.getString(FORMAT).equals(mdFormat)
&& entry.getString("layout").equals(mdLayout) && entry.getString(LAYOUT).equals(mdLayout)
&& entry.getString("interpretation").equals(mdInterpretation) && entry.getString(INTERPRETATION).equals(mdInterpretation)
&& transactions.containsKey(entry.getString("mdId"))) { && transactions.containsKey(entry.getString(MD_ID))) {
res.put(entry.getString("mdId"), transactions.get(entry.getString("mdId"))); res.put(entry.getString(MD_ID), transactions.get(entry.getString(MD_ID)));
} }
} }
@ -92,9 +144,7 @@ public class MdstoreClient implements Closeable {
private MongoCollection<Document> getColl( private MongoCollection<Document> getColl(
final MongoDatabase db, final String collName, final boolean abortIfMissing) { final MongoDatabase db, final String collName, final boolean abortIfMissing) {
if (!Iterables.contains(db.listCollectionNames(), collName)) { if (!Iterables.contains(db.listCollectionNames(), collName)) {
final String err = String final String err = String.format("Missing collection '%s' in database '%s'", collName, db.getName());
.format(
String.format("Missing collection '%s' in database '%s'", collName, db.getName()));
log.warn(err); log.warn(err);
if (abortIfMissing) { if (abortIfMissing) {
throw new RuntimeException(err); throw new RuntimeException(err);
@ -105,14 +155,31 @@ public class MdstoreClient implements Closeable {
return db.getCollection(collName); return db.getCollection(collName);
} }
private Document getMDStoreInfo(final String mdId) {
return Optional
.ofNullable(getColl(db, COLL_METADATA_MANAGER, true))
.map(metadataManager -> {
BasicDBObject query = (BasicDBObject) QueryBuilder.start(MD_ID).is(mdId).get();
log.info("querying current mdId: {}", query.toJson());
return Optional
.ofNullable(metadataManager.find(query))
.map(MongoIterable::first)
.orElseThrow(() -> new IllegalArgumentException("cannot find current mdstore id for: " + mdId));
})
.orElseThrow(() -> new IllegalStateException("missing collection " + COLL_METADATA_MANAGER));
}
public Iterable<String> listRecords(final String collName) { public Iterable<String> listRecords(final String collName) {
final MongoCollection<Document> coll = getColl(db, collName, false); return recordIterator(getColl(db, collName, false));
}
private Iterable<String> recordIterator(MongoCollection<Document> coll) {
return coll == null return coll == null
? new ArrayList<>() ? new ArrayList<>()
: () -> StreamSupport : () -> StreamSupport
.stream(coll.find().spliterator(), false) .stream(coll.find().spliterator(), false)
.filter(e -> e.containsKey("body")) .filter(e -> e.containsKey(BODY))
.map(e -> e.getString("body")) .map(e -> e.getString(BODY))
.iterator(); .iterator();
} }

View File

@ -0,0 +1,45 @@
package eu.dnetlib.dhp.common;
import java.io.Closeable;
import java.io.IOException;
import java.util.Iterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class MdstoreTx implements Iterable<String>, Closeable {
private static final Logger log = LoggerFactory.getLogger(MdstoreTx.class);
private final MdstoreClient mdstoreClient;
private final String mdId;
private final String currentId;
public MdstoreTx(MdstoreClient mdstoreClient, String mdId, String currentId) {
this.mdstoreClient = mdstoreClient;
this.mdId = mdId;
this.currentId = currentId;
}
@Override
public Iterator<String> iterator() {
return mdstoreClient.mdStoreRecords(mdId).iterator();
}
@Override
public void close() throws IOException {
mdstoreClient.readUnlock(mdId, currentId);
log.info("unlocked collection {}", currentId);
}
public String getMdId() {
return mdId;
}
public String getCurrentId() {
return currentId;
}
}

View File

@ -11,6 +11,8 @@ import org.bson.Document;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import com.mongodb.MongoClient;
import com.mongodb.MongoClientURI;
import com.mongodb.client.MongoCollection; import com.mongodb.client.MongoCollection;
import eu.dnetlib.dhp.aggregation.common.AggregatorReport; import eu.dnetlib.dhp.aggregation.common.AggregatorReport;
@ -46,7 +48,8 @@ public class MDStoreCollectorPlugin implements CollectorPlugin {
.orElseThrow(() -> new CollectorException(String.format("missing parameter '%s'", MDSTORE_ID))); .orElseThrow(() -> new CollectorException(String.format("missing parameter '%s'", MDSTORE_ID)));
log.info("mdId: {}", mdId); log.info("mdId: {}", mdId);
final MdstoreClient client = new MdstoreClient(mongoBaseUrl, dbName); final MongoClient mongoClient = new MongoClient(new MongoClientURI(mongoBaseUrl));
final MdstoreClient client = new MdstoreClient(mongoClient, dbName);
final MongoCollection<Document> mdstore = client.mdStore(mdId); final MongoCollection<Document> mdstore = client.mdStore(mdId);
long size = mdstore.count(); long size = mdstore.count();

View File

@ -91,6 +91,11 @@
<groupId>org.mongodb</groupId> <groupId>org.mongodb</groupId>
<artifactId>mongo-java-driver</artifactId> <artifactId>mongo-java-driver</artifactId>
</dependency> </dependency>
<dependency>
<groupId>io.fares.junit.mongodb</groupId>
<artifactId>mongodb-junit-test</artifactId>
<scope>test</scope>
</dependency>
<dependency> <dependency>
<groupId>dom4j</groupId> <groupId>dom4j</groupId>
<artifactId>dom4j</artifactId> <artifactId>dom4j</artifactId>

View File

@ -7,17 +7,20 @@ import java.util.Map;
import java.util.Map.Entry; import java.util.Map.Entry;
import org.apache.commons.io.IOUtils; import org.apache.commons.io.IOUtils;
import org.apache.commons.logging.Log; import org.slf4j.Logger;
import org.apache.commons.logging.LogFactory; import org.slf4j.LoggerFactory;
import com.mongodb.MongoClient;
import com.mongodb.MongoClientURI;
import eu.dnetlib.dhp.application.ArgumentApplicationParser; import eu.dnetlib.dhp.application.ArgumentApplicationParser;
import eu.dnetlib.dhp.common.MdstoreClient; import eu.dnetlib.dhp.common.MdstoreClient;
import eu.dnetlib.dhp.common.MdstoreTx;
import eu.dnetlib.dhp.oa.graph.raw.common.AbstractMigrationApplication; import eu.dnetlib.dhp.oa.graph.raw.common.AbstractMigrationApplication;
public class MigrateMongoMdstoresApplication extends AbstractMigrationApplication public class MigrateMongoMdstoresApplication extends AbstractMigrationApplication implements Closeable {
implements Closeable {
private static final Log log = LogFactory.getLog(MigrateMongoMdstoresApplication.class); private static final Logger log = LoggerFactory.getLogger(MigrateMongoMdstoresApplication.class);
private final MdstoreClient mdstoreClient; private final MdstoreClient mdstoreClient;
@ -38,31 +41,38 @@ public class MigrateMongoMdstoresApplication extends AbstractMigrationApplicatio
final String hdfsPath = parser.get("hdfsPath"); final String hdfsPath = parser.get("hdfsPath");
try (MigrateMongoMdstoresApplication app = new MigrateMongoMdstoresApplication(hdfsPath, mongoBaseUrl, final MongoClient mongoClient = new MongoClient(new MongoClientURI(mongoBaseUrl));
try (MigrateMongoMdstoresApplication app = new MigrateMongoMdstoresApplication(hdfsPath, mongoClient,
mongoDb)) { mongoDb)) {
app.execute(mdFormat, mdLayout, mdInterpretation); app.execute(mdFormat, mdLayout, mdInterpretation);
} }
} }
public MigrateMongoMdstoresApplication( public MigrateMongoMdstoresApplication(
final String hdfsPath, final String mongoBaseUrl, final String mongoDb) throws Exception { final String hdfsPath, final MongoClient mongoClient, final String mongoDb) throws Exception {
super(hdfsPath); super(hdfsPath);
this.mdstoreClient = new MdstoreClient(mongoBaseUrl, mongoDb); this.mdstoreClient = new MdstoreClient(mongoClient, mongoDb);
} }
public void execute(final String format, final String layout, final String interpretation) { public void execute(final String format, final String layout, final String interpretation) throws IOException {
final Map<String, String> colls = mdstoreClient.validCollections(format, layout, interpretation); final Map<String, String> colls = mdstoreClient.validCollections(format, layout, interpretation);
log.info("Found " + colls.size() + " mdstores"); log.info("Found {} mdstores", colls.size());
for (final Entry<String, String> entry : colls.entrySet()) { for (final Entry<String, String> entry : colls.entrySet()) {
log.info("Processing mdstore " + entry.getKey() + " (collection: " + entry.getValue() + ")"); log.info("Processing mdstore {}", entry.getKey());
final String currentColl = entry.getValue();
for (final String xml : mdstoreClient.listRecords(currentColl)) { final String mdID = entry.getKey();
try (final MdstoreTx tx = mdstoreClient.readLock(mdID)) {
log.info("locked collection {}", tx.getCurrentId());
for (final String xml : tx) {
emit(xml, String.format("%s-%s-%s", format, layout, interpretation)); emit(xml, String.format("%s-%s-%s", format, layout, interpretation));
} }
} }
} }
}
@Override @Override
public void close() throws IOException { public void close() throws IOException {

View File

@ -0,0 +1,75 @@
package eu.dnetlib.dhp.oa.graph.raw;
import com.mongodb.client.MongoDatabase;
import eu.dnetlib.dhp.common.MdstoreClient;
import io.fares.junit.mongodb.MongoExtension;
import io.fares.junit.mongodb.MongoForAllExtension;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.StringUtils;
import org.bson.Document;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
import org.junit.jupiter.api.io.TempDir;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
public class MigrateMongoMdstoresApplicationTest {
private static final Logger log = LoggerFactory.getLogger(MigrateMongoMdstoresApplicationTest.class);
public static final String COLL_NAME = "9eed8a4d-bb41-47c3-987f-9d06aee0dec0::1453898911558";
@RegisterExtension
public static MongoForAllExtension mongo = MongoForAllExtension.defaultMongo();
@BeforeAll
public static void setUp() throws IOException {
MongoDatabase db = mongo.getMongoClient().getDatabase(MongoExtension.UNIT_TEST_DB);
db.getCollection(COLL_NAME).insertOne(Document.parse(read("mdstore_record.json")));
db.getCollection("metadata").insertOne(Document.parse(read("mdstore_metadata.json")));
db.getCollection("metadataManager").insertOne(Document.parse(read("mdstore_metadataManager.json")));
}
@Test
public void test_MdstoreClient() throws IOException {
try(MdstoreClient client = new MdstoreClient(mongo.getMongoClient(), MongoExtension.UNIT_TEST_DB)) {
for (String xml : client.listRecords(COLL_NAME)) {
Assertions.assertTrue(StringUtils.isNotBlank(xml));
}
}
}
@Test
public void test_MigrateMongoMdstoresApplication(@TempDir Path tmpPath) throws Exception {
final String seqFile = "test_records.seq";
Path outputPath = tmpPath.resolve(seqFile);
try (MigrateMongoMdstoresApplication app = new MigrateMongoMdstoresApplication(
outputPath.toString(),
mongo.getMongoClient(),
MongoExtension.UNIT_TEST_DB)) {
app.execute("oai_dc", "store", "native");
}
Assertions.assertTrue(
Files.list(tmpPath)
.filter(f -> seqFile.contains(f.getFileName().toString()))
.findFirst()
.isPresent());
}
private static String read(String filename) throws IOException {
return IOUtils.toString(MigrateMongoMdstoresApplicationTest.class.getResourceAsStream(filename));
}
}

View File

@ -0,0 +1,8 @@
{
"_id" : ObjectId("53cf633a8274eb9536b614de"),
"mdId" : "9eed8a4d-bb41-47c3-987f-9d06aee0dec0_TURTdG9yZURTUmVzb3VyY2VzL01EU3RvcmVEU1Jlc291cmNlVHlwZQ==",
"format" : "oai_dc",
"layout" : "store",
"interpretation" : "native",
"size" : 1
}

View File

@ -0,0 +1 @@
{ "_id" : ObjectId("53cf63678274eb9536b61940"), "mdId" : "9eed8a4d-bb41-47c3-987f-9d06aee0dec0_TURTdG9yZURTUmVzb3VyY2VzL01EU3RvcmVEU1Jlc291cmNlVHlwZQ==", "currentId" : "9eed8a4d-bb41-47c3-987f-9d06aee0dec0::1453898911558", "expiring" : [ { "id" : "9eed8a4d-bb41-47c3-987f-9d06aee0dec0::1453898911558", "lastRead" : ISODate("2021-04-30T03:34:29.699Z") } ], "transactions" : [ ] }

View File

@ -0,0 +1,6 @@
{
"id" : "od________76::c6e4a36099aba4b0c390a365251428c9",
"originalId" : null,
"body" : "<?xml version=\"1.0\" encoding=\"UTF-8\"?><oai:record xmlns:dri=\"http://www.driver-repository.eu/namespace/dri\" xmlns=\"http://namespace.openaire.eu/\" xmlns:xsi=\"http://www.w3.org/2001/XMLSchema-instance\" xmlns:oai=\"http://www.openarchives.org/OAI/2.0/\" xmlns:dc=\"http://purl.org/dc/elements/1.1/\" xmlns:dr=\"http://www.driver-repository.eu/namespace/dr\" xmlns:oaf=\"http://namespace.openaire.eu/oaf\" xmlns:prov=\"http://www.openarchives.org/OAI/2.0/provenance\"><oai:header status=\"deleted\"><dri:objIdentifier>od________76::c6e4a36099aba4b0c390a365251428c9</dri:objIdentifier><dri:recordIdentifier>oai:DiVA.org:du-10007</dri:recordIdentifier><dri:dateOfCollection>2016-01-27T12:48:31.609Z</dri:dateOfCollection><dri:repositoryId>dc42663d-5257-4c6f-bf09-53cf47e36fed_UmVwb3NpdG9yeVNlcnZpY2VSZXNvdXJjZXMvUmVwb3NpdG9yeVNlcnZpY2VSZXNvdXJjZVR5cGU=</dri:repositoryId><oaf:datasourceprefix>od________76</oaf:datasourceprefix><identifier xmlns=\"http://www.openarchives.org/OAI/2.0/\">oai:DiVA.org:du-10007</identifier><datestamp xmlns=\"http://www.openarchives.org/OAI/2.0/\">2012-04-20T14:05:25Z</datestamp><setSpec xmlns=\"http://www.openarchives.org/OAI/2.0/\">HumanitiesTheology</setSpec><setSpec xmlns=\"http://www.openarchives.org/OAI/2.0/\">du</setSpec><setSpec xmlns=\"http://www.openarchives.org/OAI/2.0/\">studentThesis</setSpec></oai:header></oai:record>",
"timestamp" : NumberLong("1453898911613")
}

View File

@ -412,6 +412,11 @@
<artifactId>mongo-java-driver</artifactId> <artifactId>mongo-java-driver</artifactId>
<version>${mongodb.driver.version}</version> <version>${mongodb.driver.version}</version>
</dependency> </dependency>
<dependency>
<groupId>io.fares.junit.mongodb</groupId>
<artifactId>mongodb-junit-test</artifactId>
<version>1.1.0</version>
</dependency>
<dependency> <dependency>
<groupId>org.postgresql</groupId> <groupId>org.postgresql</groupId>
<artifactId>postgresql</artifactId> <artifactId>postgresql</artifactId>