TinkerPop’s Hadoop-Gremlin
JanusGraph with TinkerPop’s Hadoop-Gremlin

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

版本配置

• 存储后端：hbase（本地）。索引后端：无。
• janusgraph：janusgraph-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文件。
• 数据存放在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 Oct 31 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 Oct 31 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 stage 5.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()