janusgraph基本使用 - 好文

TinkerPop’s Hadoop-Gremlin
<https://link.jianshu.com?t=http://tinkerpop.apache.org/docs/3.2.3/reference/#bulkloadervertexprogram>
JanusGraph with TinkerPop’s Hadoop-Gremlin
<https://link.jianshu.com?t=http://docs.janusgraph.org/latest/hadoop-tp3.html>

*
利用Hadoop-Gremlin批量导入json数据到JanusGraph中，并且比较IncrementalBulkLoader和OneTimeBulkLoader的不同。
版本配置

* 存储后端：hbase（本地）。索引后端：无。
* n-0.2.0-hadoop2版本
前期准备

Schema定义

* Schema定义写在文件test-janusgraph-schema.groovy中def
defineGratefulDeadSchema(janusGraph) { m = janusGraph.openManagement()
//人信息节点label person = m.makeVertexLabel("person").make() //properties
//使用IncrementBulkLoader导入时，去掉下面注释 //blid =
m.makePropertyKey("bulkLoader.vertex.id").dataType(Long.class).make() birth =
m.makePropertyKey("birth").dataType(Date.class).make() age = m.makePropertyKey(
"age").dataType(Integer.class).make() name = m.makePropertyKey("name").dataType(
String.class).make() //index index = m.buildIndex("nameCompositeIndex",
Vertex.class).addKey(name).unique().buildCompositeIndex()
//使用IncrementBulkLoader导入时，去掉下面注释 //bidIndex =
m.buildIndex("byBulkLoaderVertexId",
Vertex.class).addKey(blid).indexOnly(person).buildCompositeIndex() m.commit() }
满足Schema格式的json文件

* 我的json文件怎么来的？将graph导出成json文件
<https://link.jianshu.com?t=http://tinkerpop.apache.org/docs/3.2.3/reference/#graphson-reader-writer>
。
* 数据存放在test-modern.json中{"id":4136,"label":"person","properties":{"name":[{
"id":"16t-36w-5j9","value":"lisi"}],"birth":[{"id":"1z9-36w-3yd","value":
1509443638951}],"age":[{"id":"101-26w-5qt","value":4136}]}} {"id":4702,"label":
"person","properties":{"name":[{"id":"171-38o-5j9","value":"fu1 "}],"birth":[{
"id":"1zh-38o-3yd","value":1509043638952}],"age":[{"id":"1l9-38o-4qt","value":1
}]}} {"id":4700,"label":"person","properties":{"name":[{"id":"171-38o-5j9",
"value":"fu2 "}],"birth":[{"id":"1zh-38o-3yd","value":1509043638976}],"age":[{
"id":"1l9-38o-4qt","value":1}]}}
目标图的配置

* 下面只列了定制的配置，默认配置没显示。
* 配置信息在janusgraph-test.properties文件中storage.backend=hbase schema.default =
none# true：在批量导入或api添加时，会进行一致性校验，否则不会进行 #
本例子中的一致性：在name属性上建立了唯一索引，所以name不允许有重复值。 storage.batch-loading=true
HadoopGraph图的配置

* 保存在hadoop-graphson.properties文件中
* 其中inputLocation指定了需要导入的json文件路径。（相对于janusgraph-0.2.0-hadoop2的路径）
gremlin.graph=org.apache.tinkerpop.gremlin.hadoop.structure.HadoopGraph
gremlin.hadoop.graphInputFormat=org.apache.tinkerpop.gremlin.hadoop.structure.io.graphson.GraphSONInputFormat
gremlin.hadoop.graphOutputFormat=org.apache.hadoop.mapreduce.lib.output.NullOutputFormat
gremlin.hadoop.inputLocation=./data/zl/test-modern.json
gremlin.hadoop.outputLocation=output gremlin.hadoop.jarsInDistributedCache=true
##################################### # GiraphGraphComputer Configuration #
##################################### giraph.minWorkers=2 giraph.maxWorkers=2
giraph.useOutOfCoreGraph=true giraph.useOutOfCoreMessages=true
mapred.map.child.java.opts=-Xmx1024m mapred.reduce.child.java.opts=-Xmx1024m
giraph.numInputThreads=4 giraph.numComputeThreads=4
giraph.maxMessagesInMemory=100000#################################### #
SparkGraphComputer Configuration # ####################################
spark.master=local[*] spark.executor.memory=1g
spark.serializer=org.apache.spark.serializer.KryoSerializer
文件位置

*

我将上述文件都放在了D:\soft\janusgraph-0.2.0-hadoop2\data\zl目录下，即janusgraph安装目录的自己创建的一个zl文件夹。

image.png
开始批量导入

* 先启动hbase。
* 以下操作都是在D:\soft\janusgraph-0.2.0-hadoop2目录下进行。
* 我使用的是Git Bash命令终端。
使用OneTimeBulkLoader批量导入

* 1.打开gremlin./bin/gremlin.bat
*
2.创建shcema，可以将下面整个粘贴到命令行中。会等待一些时间，执行完成后，在hbase中会看到多了janusgraph表，并且查询到里面有了数据（是配置以及schema数据）
:load data/zl/test-janusgraph-schema.groovy graph = JanusGraphFactory.open(
'data/zl/janusgraph-test.properties') defineGratefulDeadSchema(graph)
* 3.使用OneTimeBulkLoader批量导入，导入完成后hbase中又会多出3行数据（json文件中数据已经被到入成功了）graph =
GraphFactory.open('data/zl/hadoop-graphson.properties') blvp =
BulkLoaderVertexProgram.build().bulkLoader(OneTimeBulkLoader).writeGraph(
'data/zl/janusgraph-test.properties').create(graph)
graph.compute(SparkGraphComputer).program(blvp).submit().get()
* 4.查看导入的节点数据graph = JanusGraphFactory.open(
'data/zl/janusgraph-test.properties') g = graph.traversal() g.V().valueMap()
查到的数据类似下面结构
==>[name:[lisi],birth:[Tue Oct 31 17:53:58 CST 2017],age:[10000]]
==>[name:[zhouliang],birth:[Tue Oct31 17:53:58 CST 2017],age:[10000]]
* 5.重复执行第4步，都会导入成功，你会发现最后数据重复了。
*
6.此时修改janusgraph-test.properties中的storage.batch-loading=false，再次执行第4步，就会报错，提示有唯一性约束冲突，类似下面提示信息
org.janusgraph.core.SchemaViolationException: Adding this property for key
[name] and value [lisi] violates a uniqueness constraint [nameCompositeIndex]
使用IncrementBulkLoader批量导入

* 首先停掉gremlin console
* 删除hbase中的janusgraph表
* 将test-janusgraph-schema.groovy文件中的注释代码去掉。
* 在janusgraph-test.properties设置storage.batch-loading=true。
* 然后以下操作都是在D:\soft\janusgraph-0.2.0-hadoop2目录下进行。
* 我使用的是Git Bash命令终端。
* 1.打开gremlin./bin/gremlin.bat
*
2.创建shcema，可以将下面整个粘贴到命令行中。会等待一些时间，执行完成后，在hbase中会看到多了janusgraph表，并且查询到里面有了数据（是配置以及schema数据），因为schema多了两行代码，所以数据也会多几行。
:load data/zl/test-janusgraph-schema.groovy graph = JanusGraphFactory.open(
'data/zl/janusgraph-test.properties') defineGratefulDeadSchema(graph)
* 3.使用IncrementBulkLoader批量导入，导入完成后hbase中又会多出3行数据（json文件中数据已经被到入成功了）graph =
GraphFactory.open('data/zl/hadoop-graphson.properties') blvp =
BulkLoaderVertexProgram.build().writeGraph('data/zl/janusgraph-test.properties'
).create(graph) graph.compute(SparkGraphComputer).program(blvp).submit().get()
* 4.查看导入的节点数据graph = JanusGraphFactory.open(
'data/zl/janusgraph-test.properties') g = graph.traversal() g.V().valueMap()
* 查到的数据类似下面结构，注意，有bulkLoader.vertex.id属性，且其值为json文件中顶点的id值。
==>[name:[fu1],birth:[Tue Oct31 17:53:58 CST 2017],bulkLoader.vertex.id:[4702
],age:[10000]] ==>[name:[lisi],birth:[Tue Oct 31 17:53:58 CST 2017
],bulkLoader.vertex.id:[4136],age:[10000]]
* 如果在schema中没定义bulkLoader.vertex.id属性就会报错，类似如下：Undefined type used in query:
bulkLoader.vertex.id
* 如果没有定义byBulkLoaderVertexId索引会有警告提示，类似如下：16:35:41 WARN org.janusgraph.graphdb
.transaction.StandardJanusGraphTx - Query requires iterating over all vertices
[(~label = person AND bulkLoader.vertex.id = 4136)]. For better performance, use
indexes
*
5.重复执行第4步，都会导入成功，你会发现最后数据没有多，还是原来数据。

*

6.修改test-modern.json文件中顶点属性值，或修改顶点id值，然后在执行第4步，会发现json中id变的顶点会再次添加成功；id没变的顶点，但是其属性值变了，最后到图中对应的顶点的属性值也变化了。（其中如果json添加了新的属性，图中也会添加新的属性，但是json中属性变少了，图中的对应属性还在）。

*
7.修改json文件，将某两个顶点的id设置成相同的，在执行第4步，会报错，类似如下：
16:18:04 WARN org.apache.spark.scheduler.TaskSetManager - Lost task 0.0 in
stage5.0 (TID 3, localhost): java.lang.IllegalStateException: The property does
not exist as the key has no associated value for the provided element: v[4136]
:bulkLoader.vertex.id at
org.apache.tinkerpop.gremlin.structure.Property$Exceptions.propertyDoesNotExist(Property.
java:155)
*
8.此时修改janusgraph-test.properties中的storage.batch-loading=false，再次执行第4步，如果导入json数据的name和图中的已存在的name值有冲突就会报错，提示有唯一性约束冲突，类似下面提示信息。
org.janusgraph.core.SchemaViolationException: Adding this property for key
[name] and value [lisi] violates a uniqueness constraint [nameCompositeIndex]
结论

* storage.batch-loading设置为false，在批量导入和api添加节点时都会进行一致性校验；否则不会进行校验。
* OneTimeBulkLoader：一次批量导入数据，不会保存源图（此案例是json）中的id，导入数据不会开启事务。
*
IncrementBulkLoader：增量导入数据，并且通过bulkLoader.vertex.id属性保存源图中的id值，对于id已导入过数据会执行更新操作。为此每导入一个顶点数据都会执行如下逻辑：获取要导入顶点的id值，查询图中是否有某个顶点的bulkLoader.vertex.id值等于id值的，如果等于，则使用要插入的值，更新该图中已存在的顶点属性；如果不存在，则直接添加。
* 之所以在schema中添加下面代码，就是加快查询速度。bidIndex = m.buildIndex("byBulkLoaderVertexId",
Vertex.class).addKey(blid).indexOnly(person).buildCompositeIndex()

作者：zlcook
链接：https://www.jianshu.com/p/68117c2082a9
來源：简书
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。

热门工具换一换