jenkins-bot has submitted this change and it was merged.
Change subject: add console-based loading
......................................................................
add console-based loading
Usage:
```
g = TitanFactory.open("config.properties")
loader = tool.DumpLoader.local(g)
loader.latestDumpName("20150126.json.gz").load()
```
Change-Id: I7137acd378e628940991821a051524d25c72de86
---
M runit.groovy
M runit.sh
M src/main/groovy/org/wikidata/gremlin/Schema.groovy
M src/main/groovy/org/wikidata/gremlin/tool/DumpLoader.groovy
M src/main/java/org/wikidata/gremlin/LoadingTraversal.java
5 files changed, 133 insertions(+), 33 deletions(-)
Approvals:
Manybubbles: Looks good to me, approved
jenkins-bot: Verified
diff --git a/runit.groovy b/runit.groovy
index 4539638..bc04503 100644
--- a/runit.groovy
+++ b/runit.groovy
@@ -13,10 +13,4 @@
* limitations under the License.
*/
g = TitanFactory.open("config.properties")
-groovy.grape.Grape.grab(group:'org.wikidata', module:'gremlin',
version:'0.0.9-SNAPSHOT')
-import org.wikidata.gremlin.*;
-w = org.wikidata.gremlin.ConsoleInit.init(this)
-propLoader = new org.wikidata.gremlin.DataLoader(g, true).failOnError(true)
-dataLoader = new org.wikidata.gremlin.DataLoader(g)
-//batchLoader = new org.wikidata.gremlin.DataLoader(g,
true).batch(true).failOnError(true)
-q = new org.wikidata.gremlin.QueryEngine(g)
+loader = tool.DumpLoader.local(g)
diff --git a/runit.sh b/runit.sh
index 23a2320..63a0326 100644
--- a/runit.sh
+++ b/runit.sh
@@ -13,6 +13,4 @@
# limitations under the License.
#
-export INSTANCE=$1
-rm ~/.groovy/grapes/org.wikidata/gremlin/jars/gremlin-*-SNAPSHOT.jar
sh bin/gremlin.sh -l INFO runit.groovy
diff --git a/src/main/groovy/org/wikidata/gremlin/Schema.groovy
b/src/main/groovy/org/wikidata/gremlin/Schema.groovy
index e552158..f42fca2 100644
--- a/src/main/groovy/org/wikidata/gremlin/Schema.groovy
+++ b/src/main/groovy/org/wikidata/gremlin/Schema.groovy
@@ -52,7 +52,6 @@
def lref = addVertexLabel(mgmt, 'reference') // Reference node
// properties
def wikibaseId = addProperty(mgmt, 'wikibaseId', String.class)
- def vid = addProperty(mgmt, 'vid', String.class)
def type = addProperty(mgmt, 'type', String.class)
def datatype = addProperty(mgmt, 'datatype', String.class)
addProperty(mgmt, 'modified', Long.class)
@@ -73,19 +72,18 @@
// Labels
def claims = addUnidirectedEdgeLabel(mgmt, 'claim', prop, wikibaseId,
hash, datatype)
- def refs = addUnidirectedEdgeLabel(mgmt, 'reference', wikibaseId, hash,
etype, prop)
+ def refs = addEdgeLabel(mgmt, 'reference', wikibaseId, hash, etype)
// Regular indexes
addIndex(mgmt, 'by_wikibaseId', Vertex.class, [wikibaseId], true)
addIndex(mgmt, 'by_wikibaseIdE', Edge.class, [wikibaseId])
- addIndex(mgmt, 'by_vid', Vertex.class, [vid], true)
addIndex(mgmt, 'by_specialValueNode', Vertex.class, [specialValueNode],
true)
addIndex(mgmt, 'by_type', Vertex.class, [type])
addIndex(mgmt, 'by_etype', Edge.class, [etype])
addIndex(mgmt, 'by_prop', Edge.class, [prop])
//addIndex(mgmt, 'by_hash', Vertex.class, [hash]) //? may not be needed
addIndex(mgmt, 'by_Ehash', Edge.class, [hash])
- // edges by badges
+ // vertices by badges
addIndex(mgmt, 'by_badge', Vertex.class, [badge])
// Vertex-centric indexes
diff --git a/src/main/groovy/org/wikidata/gremlin/tool/DumpLoader.groovy
b/src/main/groovy/org/wikidata/gremlin/tool/DumpLoader.groovy
index baaa853..515090e 100644
--- a/src/main/groovy/org/wikidata/gremlin/tool/DumpLoader.groovy
+++ b/src/main/groovy/org/wikidata/gremlin/tool/DumpLoader.groovy
@@ -16,6 +16,7 @@
import com.tinkerpop.gremlin.driver.Client
import com.tinkerpop.gremlin.driver.Cluster
+import com.tinkerpop.gremlin.structure.Graph
import groovy.json.JsonSlurper
import groovy.util.CliBuilder
import groovy.util.logging.Slf4j
@@ -79,6 +80,12 @@
System.exit(0)
}
+ static DumpLoader local(Graph g)
+ {
+ def loader = new DumpLoader()
+ loader.sync(new LocalSync(g).ensureSchema())
+ }
+
/**
* Name of the latest dump file.
*/
@@ -116,6 +123,7 @@
def latestDumpName(String latestDumpName) {
this.latestDumpName = latestDumpName
+ dumpDate = null
this
}
@@ -141,9 +149,20 @@
def load() {
if (skipToLine < 0) {
- loadLanguages()
- loadSites()
- loadProperties()
+ if(!sync.getVar('languagesLoaded')) {
+ loadLanguages()
+ sync.setVar('languagesLoaded', true)
+ }
+ if(!sync.getVar('sitesLoaded')) {
+ loadSites()
+ sync.setVar('sitesLoaded', true)
+ }
+ // TODO: this means properties are loaded only once, which
+ // may not work for partial/incremental dumps
+ if(!sync.getVar('propsLoaded')) {
+ loadProperties()
+ sync.setVar('propsLoaded', true)
+ }
}
loadItems()
}
@@ -156,6 +175,7 @@
// Loading the languages in a big call causes timeouts. This doesn't.
def loaded = new
JsonSlurper().parse(getClass().getResource("/languages.json"), 'UTF-8')
loaded?.rows.each({sync.language(it[0])})
+
}
/**
@@ -173,7 +193,7 @@
log.info "Loading properties"
eachEntity(latestProperties(), 0, 'load properties',
{it.startsWith('{')},
- {line, lineNumber -> sync.entity('load properties', line, lineNumber)})
+ {line, lineNumber -> sync.entity('load properties', line, lineNumber,
true)})
sync.sync('load properties', null)
log.info "Resume now possible with -s <line number>. `-s 0` will resume
from this point. See line number logs for future points."
}
@@ -184,14 +204,14 @@
private def loadItems() {
log.info "Loading items"
eachEntity(latestDump(), skipToLine, 'load items',
- {it.startsWith('{') && !it.startsWith('{"id":"P')},
- {line, lineNumber -> sync.entity('load items', line, lineNumber)})
+ {it.startsWith('{')},
+ {line, lineNumber -> sync.entity('load items', line, lineNumber, false)})
sync.sync('load items', null)
log.info "Recording last update time"
if (dumpDate == null) {
dumpDate = latestDumpName.substring(0, 8)
}
- sync.lastUpdate(FILE_NAME_FORMAT.parseDateTime(dumpDate))
+ sync.setVar('lastUpdate',
FILE_NAME_FORMAT.parseDateTime(dumpDate).getMillis())
}
private def latestProperties() {
@@ -248,6 +268,25 @@
}
/**
+ * Return gzip stream or regular stream depending on file contents
+ * @param src
+ * @return InputStream
+ */
+ private def getMaybeGzipStream(InputStream src) {
+ // check if it's gzip for non-gzip support too
+ def pbs = new PushbackInputStream(src, 2)
+ byte[] gzipSig = new byte[2]
+ pbs.read(gzipSig)
+ pbs.unread(gzipSig)
+ if((gzipSig[0] == (byte) GZIPInputStream.GZIP_MAGIC)
+ && (gzipSig[1] == (byte) (GZIPInputStream.GZIP_MAGIC >> 8))) {
+ new GZIPInputStream(pbs)
+ } else {
+ pbs
+ }
+ }
+
+ /**
* Opens an input stream that reads the dump as text, unzipping it on the
fly.
* If the dump isn't downloaded then it downloads it completely and then
* reopens it as a stream.
@@ -287,7 +326,8 @@
latestDumpPart.renameTo(latestDump)
}
}
- new GZIPInputStream(latestDump.newInputStream())
+ // check if it's gzip for non-gzip support too
+ getMaybeGzipStream(latestDump.newInputStream())
}
void logError(String message, Throwable t) {
@@ -298,9 +338,10 @@
interface Sync {
void language(language)
void site(site)
- void entity(operationName, entity, lineNumber)
+ void entity(operationName, entity, lineNumber, loadProperties)
void sync(operationName, lineNumber)
- void lastUpdate(lastUpdate)
+ def getVar(name)
+ void setVar(name, value)
}
static class PrintSync implements Sync {
@@ -311,11 +352,12 @@
void site(site) {
print "Site ${site}"
}
- void entity(operationName, blob, lineNumber) {
+ void entity(operationName, blob, lineNumber, loadProperties) {
println json.parseText(blob).id
}
void sync(operationName, lineNumber) {}
- void lastUpdate(lastUpdate) {}
+ def getVar(name) { null }
+ void setVar(name, value) {}
}
static class GremlinSync implements Sync {
@@ -353,14 +395,15 @@
}
}
- void entity(operationName, blob, lineNumber) {
+ void entity(operationName, blob, lineNumber, loadProperties) {
// Sending a big String over Kryo wasn't working so we parse it and send
it as json
// which should work better. Its not as efficient but it works.
def entity = new JsonSlurper().parseText(blob)
def task = {loader.retryMostErrors("updating ${entity.id}"){
- client.submitAsync("g.wt().loadFromEntity(entity);
g.commitIfSupported()",
- [entity: entity])
+ client.submitAsync("g.wt().loadFromEntity(entity, props);
g.commitIfSupported()",
+ [entity: entity, props: loadProperties])
}}
+
// The first few items and all properties will have lock contention
issues if we
// use too many threads. So we single thread them.
if (entity.type != 'item' || lineNumber < 500) {
@@ -384,10 +427,19 @@
}
}
- void lastUpdate(lastUpdate) {
- loader.retryMostErrors('updating last update time'){
- client.submitAsync("g.variables().set('lastUpdate', lastUpdate)",
[lastUpdate: lastUpdate.getMillis()])
+ void setVar(name, value) {
+ loader.retryMostErrors("setting variable $name"){
+ client.submitAsync("g.variables().set(name, value)", [name: name,
value: value])
}
+
+ }
+
+ def getVar(name) {
+ loader.retryMostErrors("fetching variable $name"){
+ def res = client.submit("g.variables().get(name).orElse(null)", [name:
name]).all().get()
+ res[0].getObject()
+ }
+ null
}
void close() {
@@ -396,4 +448,54 @@
futures.each{it.get()}
}
}
+
+ static class LocalSync implements Sync {
+ private def g
+ private def eloader
+ private def ploader
+
+ public LocalSync(Graph g) {
+ this.g = g
+ this.eloader = g.wt().loader()
+ this.ploader = g.wt().buildLoader(true) // property loader
+ }
+
+ LocalSync ensureSchema() {
+ g.wt().ensureSchema()
+ this
+ }
+
+ void language(language) {
+ eloader.initLanguage(language)
+ this
+ }
+
+ void site(site) {
+ eloader.initSite(site)
+ }
+
+ void entity(operationName, blob, lineNumber, loadProperties) {
+ def entity = new JsonSlurper().parseText(blob)
+ if(loadProperties) {
+ ploader.loadFromItem(entity)
+ } else {
+ eloader.loadFromItem(entity)
+ }
+ g.commitIfSupported()
+ }
+
+ void sync(operationName, lineNumber) {}
+
+ def getVar(name) {
+ g.variables().get(name).orElse(null)
+ }
+
+ void setVar(name, value) {
+ g.variables().set(name, value)
+ g.commitIfSupported()
+ }
+
+ void close() {}
+ }
+
}
diff --git a/src/main/java/org/wikidata/gremlin/LoadingTraversal.java
b/src/main/java/org/wikidata/gremlin/LoadingTraversal.java
index fd25dd5..4e64d66 100644
--- a/src/main/java/org/wikidata/gremlin/LoadingTraversal.java
+++ b/src/main/java/org/wikidata/gremlin/LoadingTraversal.java
@@ -70,8 +70,8 @@
/**
* Load from a pre-fetched object.
*/
- default Self loadFromEntity(Object item) {
- return cast(asAdmin().addStep(new StartStep<>(this,
loader().loadFromItem(item))));
+ default Self loadFromEntity(Object item, boolean loadProperties) {
+ return cast(asAdmin().addStep(new StartStep<>(this,
buildLoader(loadProperties).loadFromItem(item))));
}
/**
@@ -89,4 +89,12 @@
loader.setEntitySource(graph().variables().<String>get("source").orElse(Loader.WIKIDATA_URL));
return loader;
}
+ /**
+ * Build the loader.
+ */
+ default Loader buildLoader(boolean loadProperties) {
+ Loader loader = new Loader(graph(), loadProperties);
+
loader.setEntitySource(graph().variables().<String>get("source").orElse(Loader.WIKIDATA_URL));
+ return loader;
+ }
}
--
To view, visit https://gerrit.wikimedia.org/r/188117
To unsubscribe, visit https://gerrit.wikimedia.org/r/settings
Gerrit-MessageType: merged
Gerrit-Change-Id: I7137acd378e628940991821a051524d25c72de86
Gerrit-PatchSet: 4
Gerrit-Project: wikidata/gremlin
Gerrit-Branch: master
Gerrit-Owner: Smalyshev <[email protected]>
Gerrit-Reviewer: Manybubbles <[email protected]>
Gerrit-Reviewer: Smalyshev <[email protected]>
Gerrit-Reviewer: jenkins-bot <>
_______________________________________________
MediaWiki-commits mailing list
[email protected]
https://lists.wikimedia.org/mailman/listinfo/mediawiki-commits
