摘要： 主要介绍如何通过官方 ETL 工具 Exchange 将业务线上数据从 Neo4j 直接导入到 Nebula Graph 以及在导入过程中遇到的问题和优化方法。
本文首发于 Nebula 论坛：https://discuss.nebula-graph.com.cn/t/topic/2044
随着业务数据量不断增长，业务对图数据库在线数据实时更新写入和查询的效率要求也不断增加。Neo4j 存在明显性能不足，Neo4j 社区开源版本只支持单机部署，扩展能力存在比较大的问题，无法满足读写性能的线性扩展以及读写分离的业务需求，并且开源版本 Neo4j 对点和边的总数据量也有限制；而 Neo4j 企业版因果集群也存在单机主节点 Cypher 实时写入的性能瓶颈。
相比于 Neo4j，Nebula Graph 最大的特色便是采用 shared-nothing 分布式的架构，无单主写入瓶颈问题，读写支持线性扩展，擅长处理千亿节点、万亿条边的超大规模数据集。
本文主要介绍如何通过官方 ETL 工具 Exchange 将业务线上数据从 Neo4j 直接导入到 Nebula Graph 以及在导入过程中遇到的问题和优化方法。其中绝大部分问题都已经通过论坛发帖的方式得到社区的支持和解决，本文会结合问题进行逐一列举。
系统环境：
软件环境：
注意：单台机器部署 Nebula 多节点的端口分配：每个 storage 还会将用户配置的端口号 + 1 的端口作为内部使用。请参考论坛帖子 nebula 从 neo4j 导入数据出现 Get UUID Failed 错误
根据 Neo4j 点和边的属性信息创建 Nebula Graph 的 Tag 和 Edge 结构，这里需要注意一点，业务可能会根据不同需求只在部分点和边上增加 Neo4j 点和边的属性信息，其他点和边对应的属性为 NULL，所以需要先跟业务明确一下点和边的全部属性信息，避免遗漏属性。Nebula Graph 的 Schema 信息类似 MySQL，支持 Create 和 Alter 添加属性，并且所有的 Tag 和 Edge 的元数据信息是一致的。
1 、Nebula Graph 创建 Tag 和 Edge
# 示例
# 创建图空间，10 个分区，3 个 storage 副本。
CREATE SPACE test(partition_num=10,replica_factor=3);
# 选择图空间 test
USE test;
# 创建标签 tagA
CREATE TAG tagA(vid string, field-a0 string, field-a1 bool, field-a2 double);
# 创建标签 tagB
CREATE TAG tagB(vid string, field-b0 string, field-b1 bool, field-b2 double);
# 创建边类型 edgeAB
CREATE EDGE edgeAB(vid string, field-e0 string, field-e1 bool, field-e2 double);
2 、Exchange 导入配置文件
bolt+routing 的方式连接 neo4j，如果是因果集群，可以选择一个从节点进行 bolt 方式直连读取数据，减少集群压力。{
# Spark relation config
spark: {
app: {
name: Spark Writer
}
driver: {
cores: 1
maxResultSize: 1G
}
cores {
max: 16
}
}
# Nebula Graph relation config
nebula: {
address:{
graph:["xxx.xxx.xxx.xx:3699"]
meta:["xxx.xxx.xxx.xx:45500"]
}
user: user
pswd: password
space: test
connection {
timeout: 3000
retry: 3
}
execution {
retry: 3
}
error: {
max: 32
output: /tmp/errors
}
rate: {
limit: 1024
timeout: 1000
}
}
# Processing tags
tags: [
# Loading tag from neo4j
{
name: tagA
type: {
source: neo4j
sink: client
}
server: "bolt://xxx.xxx.xxx.xxx:7687"
user: neo4j
password: neo4j
exec: "match (n:tagA) where id(n) < 300000000 return n.vid as vid, n.field-a0 as field-a0, n.field-a1 as field-a1, n.field-a2 as field-a2 order by id(n)"
fields: [vid, field-a0, field-a1, field-a2]
nebula.fields: [vid, field-a0, field-a1, field-a2]
vertex: {
field: vid
policy: "uuid"
}
partition: 10
batch: 1000
check_point_path: /tmp/test
}
# Loading tag from neo4j
{
name: tagB
type: {
source: neo4j
sink: client
}
server: "bolt://xxx.xxx.xxx.xxx:7687"
user: neo4j
password: neo4j
exec: "match (n:tagB) where id(n) < 300000000 return n.vid as vid, n.field-b0 as field-b0, n.field-b1 as field-b1, n.field-b2 as field-b2 order by id(n)"
fields: [vid, field-b0, field-b1, field-b2]
nebula.fields: [vid, field-b0, field-b1, field-b2]
vertex: {
field: vid
policy: "uuid"
}
partition: 10
batch: 1000
check_point_path: /tmp/test
}
]
# Processing edges
edges: [
# Loading edges from neo4j
{
name: edgeAB
type: {
source: neo4j
sink: client
}
server: "bolt://xxx.xxx.xxx.xxx:7687"
user: neo4j
password: neo4j
exec: "match (a:tagA)-[r:edgeAB]->(b:tagB) where id(r) < 300000000 return n.vid as vid, n.field-e0 as field-e0, n.field-e1 as field-e1, n.field-e2 as field-e2 order by id(r)"
fields: [vid, field-e0, field-e1, field-e2]
nebula.fields: [vid, field-e0, field-e1, field-e2]
source: {
field: a.vid
policy: "uuid"
}
target: {
field: b.vid
policy: "uuid"
}
partition: 10
batch: 1000
check_point_path: /tmp/test
}
]
}
3 、执行导入命令
nohup spark-submit --class com.vesoft.nebula.tools.importer.Exchange --master "local" exchange-1.1.0.jar -c test.conf > test.log &
4 、查看导入 Nebula Graph 的数据量
./bin/db_dump --space=test --db_path=./data/storage/nebula/ --meta_server=127.0.0.1:45500 -limit 0 --mode=stat --tags=tagA,tagB --edges=edgeAB
注意：Nebula 1.x 版本目前还只能用 db_dump 统计，2.0 会支持 nGQL 命令的方式统计数量。
增量数据导入主要是通过 Neo4j 内部点和边的自增 id() 进行切割，在导入配置文件 exec 项执行 Neo4j Cypher 语句时增加 id() 范围限制，但前提是需要业务停掉删数据操作，因为增量导入时，如果之前的数据被删除后 Neo4j 会复用 id()，这会导致复用 id() 的增量数据导入时查询不到造成数据丢失。当然业务如果有条件支持 Neo4j Nebula 双写的话，增量导入就不会出现这种问题。
exec: "match (n:user) where id(n) >= 300000000 and id(n) < 400000000 return xxx order by id(n)"
请参考论坛帖子 neo4j 到 nebula 如何做增量导入
使用 Exchange 导入过程中遇到两个问题，及时的得到官方 @nicole 的支持和解决，具体请参考下面两个帖子：
问题 1：Exchange 不支持「换行回车」等特殊字符的转义。如下 string 数据中带有回车，在拼接 insert 语句插入时会因为换行导致插入失败。
PR： https://github.com/vesoft-inc/nebula-java/pull/203 已经合入 exchange v1.0 分支
问题 2：Exchange 不支持属性为 NULL 的数据导入。前文 3.1 中提到，业务可能会根据不同需求为某些点和边增加属性，这时其他点和边属性则是 NULL，这样在使用 Exchange 导入时会报错。
参考帖子 2 给出的修改建议解决：修改 com.vesoft.nebula.tools.importer.processor.Processor#extraValue，增加 NULL 类型的转化值。
case NullType => {
fieldTypeMap(field) match {
case StringType => ""
case IntegerType => 0
case LongType => 0L
case DoubleType => 0.0
case BooleanType => false
}
}
关于导入效率的优化，请参考下面两个帖子：
优化 1：通过适当增加导入配置中的 partition 和 batch 值，提升导入效率。
优化 2：如果是 string 类型做 vid 的话，1.x 版本尽量使用 hash() 函数转化，2.0 版本会支持 string id 类型；如果是 int 类型做 vid 的话，可以直接使用，不用转化效率更高。
优化 3：官方建议 spark-submit 提交命令 master 配置改为 yarn-cluster， 若不使用 yarn，可配置成 spark://ip:port；我们是通过 spark-submit --master "local[16]" 的方式增加 spark 并发，导入效率比使用 "local" 提升 4 倍+，测试环境单机三节点 HDD 盘 IO 峰值能到 200-300 MB/s 。但在指定 --master "local[16]" 并发导入时遇到 hadoop 缓存问题，采用增加 hdfs 配置 fs.hdfs.impl.disable.cache=true 后重启 hadoop 解决。具体请参考第二个帖子。
使用 Exchange 从 Neo4j 导入 Nebula Graph 过程中遇到一些问题，通过积极与社区进行沟通得到了官方 @nicole 及其他小伙伴的快速响应和大力支持，这一点在 Neo4j 导入 Nebula Graph 的实践过程中起到了十分关键的作用，感谢社区的大力支持。期待支持 openCypher 的 Nebula Graph 2.0 。
这是一个专为移动设备优化的页面（即为了让你能够在 Google 搜索结果里秒开这个页面），如果你希望参与 V2EX 社区的讨论，你可以继续到 V2EX 上打开本讨论主题的完整版本。
V2EX 是创意工作者们的社区，是一个分享自己正在做的有趣事物、交流想法，可以遇见新朋友甚至新机会的地方。
V2EX is a community of developers, designers and creative people.