TinkerPop’s Hadoop-Gremlin
JanusGraph with TinkerPop’s Hadoop-Gremlin
- 利用Hadoop-Gremlin批量導(dǎo)入json數(shù)據(jù)到JanusGraph中召夹，并且比較IncrementalBulkLoader和OneTimeBulkLoader的不同递览。
版本配置
- 存儲(chǔ)后端：hbase（本地）唬党。索引后端：無波势。
- janusgraph：janusgraph-0.2.0-hadoop2版本
前期準(zhǔn)備
Schema定義
- Schema定義寫在文件test-janusgraph-schema.groovy中
def defineGratefulDeadSchema(janusGraph) {
m = janusGraph.openManagement()
//人信息節(jié)點(diǎn)label
person = m.makeVertexLabel("person").make()
//properties
//使用IncrementBulkLoader導(dǎo)入時(shí)弦蹂，去掉下面注釋
//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導(dǎo)入時(shí)敛腌，去掉下面注釋
//bidIndex = m.buildIndex("byBulkLoaderVertexId", Vertex.class).addKey(blid).indexOnly(person).buildCompositeIndex()
m.commit()
}
滿足Schema格式的json文件
- 我的json文件怎么來的封拧？將graph導(dǎo)出成json文件复唤。
- 數(shù)據(jù)存放在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}]}}
目標(biāo)圖的配置
- 下面只列了定制的配置，默認(rèn)配置沒顯示娄昆。
- 配置信息在janusgraph-test.properties文件中
storage.backend=hbase
schema.default = none
# true：在批量導(dǎo)入或api添加時(shí)佩微，會(huì)進(jìn)行一致性校驗(yàn)，否則不會(huì)進(jìn)行
# 本例子中的一致性：在name屬性上建立了唯一索引萌焰，所以name不允許有重復(fù)值哺眯。
storage.batch-loading=true
HadoopGraph圖的配置
- 保存在hadoop-graphson.properties文件中
- 其中inputLocation指定了需要導(dǎo)入的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安裝目錄的自己創(chuàng)建的一個(gè)zl文件夾奶卓。
image.png
開始批量導(dǎo)入
- 先啟動(dòng)hbase一疯。
- 以下操作都是在D:\soft\janusgraph-0.2.0-hadoop2目錄下進(jìn)行。
- 我使用的是Git Bash命令終端夺姑。
使用OneTimeBulkLoader批量導(dǎo)入
- 1.打開gremlin
./bin/gremlin.bat
- 2.創(chuàng)建shcema墩邀，可以將下面整個(gè)粘貼到命令行中。會(huì)等待一些時(shí)間盏浙，執(zhí)行完成后眉睹，在hbase中會(huì)看到多了janusgraph表，并且查詢到里面有了數(shù)據(jù)（是配置以及schema數(shù)據(jù)）
:load data/zl/test-janusgraph-schema.groovy
graph = JanusGraphFactory.open('data/zl/janusgraph-test.properties')
defineGratefulDeadSchema(graph)
- 3.使用OneTimeBulkLoader批量導(dǎo)入废膘，導(dǎo)入完成后hbase中又會(huì)多出3行數(shù)據(jù)（json文件中數(shù)據(jù)已經(jīng)被到入成功了）
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.查看導(dǎo)入的節(jié)點(diǎn)數(shù)據(jù)
graph = JanusGraphFactory.open('data/zl/janusgraph-test.properties')
g = graph.traversal()
g.V().valueMap()
查到的數(shù)據(jù)類似下面結(jié)構(gòu)
==>[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.重復(fù)執(zhí)行第4步竹海，都會(huì)導(dǎo)入成功，你會(huì)發(fā)現(xiàn)最后數(shù)據(jù)重復(fù)了殖卑。
- 6.此時(shí)修改janusgraph-test.properties中的storage.batch-loading=false站削，再次執(zhí)行第4步坊萝，就會(huì)報(bào)錯(cuò)孵稽，提示有唯一性約束沖突，類似下面提示信息
org.janusgraph.core.SchemaViolationException: Adding this property for key [name] and value [lisi] violates a uniqueness constraint [nameCompositeIndex]
使用IncrementBulkLoader批量導(dǎo)入
- 首先停掉gremlin console
- 刪除hbase中的janusgraph表
- 將test-janusgraph-schema.groovy文件中的注釋代碼去掉十偶。
- 在janusgraph-test.properties設(shè)置storage.batch-loading=true菩鲜。
- 然后以下操作都是在D:\soft\janusgraph-0.2.0-hadoop2目錄下進(jìn)行。
- 我使用的是Git Bash命令終端惦积。
- 1.打開gremlin
./bin/gremlin.bat
- 2.創(chuàng)建shcema接校，可以將下面整個(gè)粘貼到命令行中。會(huì)等待一些時(shí)間狮崩，執(zhí)行完成后蛛勉，在hbase中會(huì)看到多了janusgraph表，并且查詢到里面有了數(shù)據(jù)（是配置以及schema數(shù)據(jù)）睦柴，因?yàn)閟chema多了兩行代碼诽凌，所以數(shù)據(jù)也會(huì)多幾行。
:load data/zl/test-janusgraph-schema.groovy
graph = JanusGraphFactory.open('data/zl/janusgraph-test.properties')
defineGratefulDeadSchema(graph)
- 3.使用IncrementBulkLoader批量導(dǎo)入坦敌，導(dǎo)入完成后hbase中又會(huì)多出3行數(shù)據(jù)（json文件中數(shù)據(jù)已經(jīng)被到入成功了）
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.查看導(dǎo)入的節(jié)點(diǎn)數(shù)據(jù)
graph = JanusGraphFactory.open('data/zl/janusgraph-test.properties')
g = graph.traversal()
g.V().valueMap()
- 查到的數(shù)據(jù)類似下面結(jié)構(gòu)侣诵，注意，有bulkLoader.vertex.id屬性狱窘，且其值為json文件中頂點(diǎn)的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屬性就會(huì)報(bào)錯(cuò)，類似如下：
Undefined type used in query: bulkLoader.vertex.id
- 如果沒有定義byBulkLoaderVertexId索引會(huì)有警告提示蘸炸，類似如下：
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.重復(fù)執(zhí)行第4步躬络，都會(huì)導(dǎo)入成功，你會(huì)發(fā)現(xiàn)最后數(shù)據(jù)沒有多搭儒，還是原來數(shù)據(jù)穷当。
6.修改test-modern.json文件中頂點(diǎn)屬性值越锈，或修改頂點(diǎn)id值，然后在執(zhí)行第4步膘滨，會(huì)發(fā)現(xiàn)json中id變的頂點(diǎn)會(huì)再次添加成功甘凭；id沒變的頂點(diǎn)，但是其屬性值變了火邓，最后到圖中對應(yīng)的頂點(diǎn)的屬性值也變化了丹弱。（其中如果json添加了新的屬性，圖中也會(huì)添加新的屬性铲咨，但是json中屬性變少了躲胳，圖中的對應(yīng)屬性還在）。
7.修改json文件纤勒，將某兩個(gè)頂點(diǎn)的id設(shè)置成相同的坯苹，在執(zhí)行第4步，會(huì)報(bào)錯(cuò)摇天，類似如下：
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.此時(shí)修改janusgraph-test.properties中的storage.batch-loading=false粹湃，再次執(zhí)行第4步，如果導(dǎo)入json數(shù)據(jù)的name和圖中的已存在的name值有沖突就會(huì)報(bào)錯(cuò)泉坐，提示有唯一性約束沖突为鳄，類似下面提示信息。
org.janusgraph.core.SchemaViolationException: Adding this property for key [name] and value [lisi] violates a uniqueness constraint [nameCompositeIndex]
結(jié)論
- storage.batch-loading設(shè)置為false腕让，在批量導(dǎo)入和api添加節(jié)點(diǎn)時(shí)都會(huì)進(jìn)行一致性校驗(yàn)孤钦；否則不會(huì)進(jìn)行校驗(yàn)。
- OneTimeBulkLoader：一次批量導(dǎo)入數(shù)據(jù)纯丸，不會(huì)保存源圖（此案例是json）中的id偏形，導(dǎo)入數(shù)據(jù)不會(huì)開啟事務(wù)。
- IncrementBulkLoader：增量導(dǎo)入數(shù)據(jù)觉鼻，并且通過bulkLoader.vertex.id屬性保存源圖中的id值俊扭，對于id已導(dǎo)入過數(shù)據(jù)會(huì)執(zhí)行更新操作。為此每導(dǎo)入一個(gè)頂點(diǎn)數(shù)據(jù)都會(huì)執(zhí)行如下邏輯：獲取要導(dǎo)入頂點(diǎn)的id值滑凉，查詢圖中是否有某個(gè)頂點(diǎn)的bulkLoader.vertex.id值等于id值的统扳，如果等于，則使用要插入的值畅姊，更新該圖中已存在的頂點(diǎn)屬性咒钟；如果不存在，則直接添加若未。
- 之所以在schema中添加下面代碼朱嘴，就是加快查詢速度。
bidIndex = m.buildIndex("byBulkLoaderVertexId", Vertex.class).addKey(blid).indexOnly(person).buildCompositeIndex()
