本文主要讲述如何使用数据导入工具 Nebula Graph Exchange 将数据从 Neo4j 导入到 Nebula Graph Database。在讲述如何实操数据导入以前，咱们先来了解下 Nebula Graph 内部是如何实现这个导入功能的。java
Nebula Graph Exchange 的数据处理原理
咱们这个导入工具名字是 Nebula Graph Exchange，采用 Spark 做为导入平台，来支持海量数据的导入和保障性能。Spark 自己提供了不错的抽象——DataFrame，使得能够轻松支持多种数据源。在 DataFrame 的支持下，添加新的数据源只需提供配置文件读取的代码和返回 DataFrame 的 Reader 类，便可支持新的数据源。git
DataFrame 能够视为一种分布式存表格。DataFrame 能够存储在多个节点的不一样分区中，多个分区能够存储在不一样的机器上，从而支持并行操做。Spark 还提供了一套简洁的 API 使用户轻松操做 DataFrame 如同操做本地数据集通常。如今大多数数据库提供直接将数据导出成 DataFrame 功能，即便某个数据库并未提供此功能也能够经过数据库 driver 手动构建 DataFrame。github
Nebula Graph Exchange 将数据源的数据处理成 DataFrame 以后，会遍历它的每一行，根据配置文件中 fields 的映射关系，按列名获取对应的值。在遍历 batchSize 个行以后，Exchange 会将获取的数据一次性写入到 Nebula Graph 中。目前，Exchange 是经过生成 nGQL 语句再由 Nebula Client 异步写入数据，下一步会支持直接导出 Nebula Graph 底层存储的 sst 文件，以获取更好的性能。接下来介绍一下 Neo4j 数据源导入的具体实现。sql
Neo4j 数据导入具体实现
虽然 Neo4j 官方提供了可将数据直接导出为 DataFrame 的库，但使用它读取数据难以知足断点续传的需求，咱们未直接使用这个库，而是使用 Neo4j 官方的 driver 实现数据读取。Exchange 经过在不一样分区调取 Neo4j driver 执行不一样 skip 和 limit 的 Cypher 语句，将数据分布在不一样的分区，来获取更好的性能。这个分区数量由配置项 partition 指定。docker
Exchange 中的 Neo4jReader 类会先将用户配置中的 exec Cypher 语句，return 后边的语句替换成 count(*) 执行获取数据总量，再根据分区数计算每一个分区的起始偏移量和大小。这里若是用户配置了 check_point_path 目录，会读取目录中的文件，若是处于续传的状态，Exchange 会计算出每一个分区应该的偏移量和大小。而后每一个分区在 Cypher 语句后边添加不一样的 skip 和 limit，调用 driver 执行。最后将返回的数据处理成 DataFrame 就完成了 Neo4j 的数据导入。shell
过程以下图所示：数据库
Neo4j 数据导入实践
咱们这里导入演示的系统环境以下：微信
- cpu name：Intel(R) Xeon(R) CPU E5-2697 v3 @ 2.60GHz
- cpu cores：14
- memory size：251G
软件环境以下：架构
- Neo4j：3.5.20 社区版
- Nebula graph：docker-compose 部署，默认配置
- Spark：单机版，版本为 2.4.6 pre-build for hadoop2.7
因为 Nebula Graph 是强 schema 数据库，数据导入前需先进行建立 Space，建 Tag 和 Edge 的 schema，具体的语法能够参考这里。并发
这里建了名为 test 的 Space，副本数为 1。这里建立了两种 Tag 分别为 tagA 和 tagB，均含有 4 个属性的点类型，此外，还建立一种名为 edgeAB 的边类型，一样含有 4 个属性。具体的 nGQL 语句以下所示：
# 建立图空间 CREATE SPACE test(replica_factor=1); # 选择图空间 test USE test; # 建立标签 tagA CREATE TAG tagA(idInt int, idString string, tboolean bool, tdouble double); # 建立标签 tagB CREATE TAG tagB(idInt int, idString string, tboolean bool, tdouble double); # 建立边类型 edgeAB CREATE EDGE edgeAB(idInt int, idString string, tboolean bool, tdouble double);
同时向 Neo4j 导入 Mock 数据——标签为 tagA 和 tagB 的点，数量总共为 100 万，而且导入了链接 tagA 和 tagB 类型点边类型为 edgeAB 的边，共 1000 万个。另外须要注意的是，从 Neo4j 导出的数据在 Nebula Graph 中必须存在属性，且数据对应的类型要同 Nebula Graph 一致。
最后为了提高向 Neo4j 导入 Mock 数据的效率和 Mock 数据在 Neo4j 中的读取效率，这里为 tagA 和 tagB 的 idInt 属性建了索引。关于索引须要注意 Exchange 并不会将 Neo4j 中的索引、约束等信息导入到 Nebula Graph 中，因此须要用户在执行数据写入在 Nebula Graph 以后，自行建立索引和 REBUILD 索引（为已有数据创建索引）。
接下来就能够将 Neo4j 数据导入到 Nebula Graph 中了，首先咱们须要下载和编译打包项目，项目在 nebula-java 这个仓库下 tools/exchange 文件夹中。可执行以下命令：
git clone https://github.com/vesoft-inc/nebula-java.git cd nebula-java/tools/exchange mvn package -DskipTests
而后就能够看到 target/exchange-1.0.1.jar 这个文件。
接下来编写配置文件，配置文件的格式为：HOCON（Human-Optimized Config Object Notation），能够基于 src/main/resources/server_application.conf 文件的基础上进行更改。首先对 nebula 配置项下的 address、user、pswd 和 space 进行配置，测试环境均为默认配置，因此这里不须要额外的修改。而后进行 tags 配置，须要 tagA 和 tagB 的配置，这里仅展现 tagA 配置，tagB 和 tagA 配置相同。
{
# ======neo4j链接设置=======
name: tagA
# 必须和 Nebula Graph 的中 tag 名字一致，须要在 Nebula Graph 中事先建好 tag
server: "bolt://127.0.0.1:7687"
# neo4j 的地址配置
user: neo4j
# neo4j 的用户名
password: neo4j
# neo4j 的密码
encryption: false
# (可选): 传输是否加密，默认值为 false
database: graph.db
# (可选): neo4j database 名称，社区版不支持
# ======导入设置============
type: {
source: neo4j
# 还支持 PARQUET、ORC、JSON、CSV、HIVE、MYSQL、PULSAR、KAFKA...
sink: client
# 写入 Nebula Graph 的方式，目前仅支持 client，将来会支持直接导出 Nebula Graph 底层数据库文件
}
nebula.fields: [idInt, idString, tdouble, tboolean]
fields : [idInt, idString, tdouble, tboolean]
# 映射关系 fields，上方为 nebula 的属性名，下方为 neo4j 的属性名，一一对应
# 映射关系的配置是 List 而不是 Map，是为了保持 fields 的顺序，将来直接导出 nebula 底层存储文件时须要
vertex: idInt
# 做为 nebula vid 的 neo4j field，类型须要是整数(long or int)。
partition: 10
# 分区数
batch: 2000
# 一次写入 nebula 多少数据
check_point_path: "file:///tmp/test"
# (可选): 保存导入进度信息的目录，用于断点续传
exec: "match (n:tagA) return n.idInt as idInt, n.idString as idString, n.tdouble as tdouble, n.tboolean as tboolean order by n.idInt"
}
边的设置大部分与点的设置无异，但因为边在 Nebula Graph 中有起点的 vid 和终点的 vid 标识，因此这里须要指定做为边起点 vid 的域和做为边终点 vid 的域。
下面给出边的特别配置。
source: {
field: a.idInt
# policy: "hash"
}
# 起点的 vid 设置
target: {
field: b.idInt
# policy: "uuid"
}
# 终点的 vid 设置
ranking: idInt
# (可选): 做为 rank 的 field
partition: 1
# 这里分区数设置为 1，缘由在后边
exec: "match (a:tagA)-[r:edgeAB]->(b:tagB) return a.idInt, b.idInt, r.idInt as idInt, r.idString as idString, r.tdouble as tdouble, r.tboolean as tboolean order by id(r)"
点的 vertex 和边的 source、target 配置项下均可以设置 policy hash/uuid，它能够将类型为字符串的域做为点的 vid，经过 hash/uuid 函数将字符串映射成整数。
上面的例子因为做为点的 vid 为整数，因此并不须要 policy 的设置。hash/uuid的 区别请看这里。
Cypher 标准中若是没有 order by 约束的话就不能保证每次查询结果的排序一致，虽然看起来即使不加 order by Neo4j 返回的结果顺序也是不变的，但为了防止可能形成的导入时数据丢失，仍是强烈建议在 Cypher 语句中加入 order by，虽然这会增长导入的时间。为了提高导入效率， order by 语句最好选取有索引的属性做为排序的属性。若是没有索引，也可观察默认的排序，选择合适的排序属性以提升效率。若是默认的排序找不到规律，可使用点/关系的 ID 做为排序属性，而且将 partition 的值尽可能设小，减小 Neo4j 的排序压力，本文中边 edgeAB 的 partition 就设置为 1。
另外 Nebula Graph 在建立点和边时会将 ID 做为惟一主键，若是主键已存在则会覆盖该主键中的数据。因此假如将某个 Neo4j 属性值做为 Nebula Graph 的 ID，而这个属性值在 Neo4j 中是有重复的，就会致使“重复 ID”对应的数据有且只有一条会存入 Nebula Graph 中，其它的则会被覆盖掉。因为数据导入过程是并发地往 Nebula Graph 中写数据，最终保存的数据并不能保证是 Neo4j 中最新的数据。
这里还要留意下断点续传功能，在断点和续传之间，数据库不该该改变状态，如添加数据或删除数据，且 partition 数量也不能更改，不然可能会有数据丢失。
最后因为 Exchange 须要在不一样分区执行不一样 skip 和 limit 的 Cypher 语句，因此用户提供的 Cypher 语句不能含有 skip 和 limit 语句。
接下来就能够运行 Exchange 程序导数据了，执行以下命令：
$SPARK_HOME/bin/spark-submit --class com.vesoft.nebula.tools.importer.Exchange --master "local[10]" target/exchange-1.0.1.jar -c /path/to/conf/neo4j_application.conf
在上述这些配置下，导入 100 万个点用时 13s，导入 1000 万条边用时 213s，总用时是 226s。
附：Neo4j 3.5 Community 和 Nebula Graph 1.0.1的一些比较
Neo4j 和 Nebula Graph 在系统架构、数据模型和访问方式上都有一些差别，下表列举了常见的异同
做者有话说：Hi，我是李梦捷，图数据库 Nebula Graph 的研发工程师，若是你对此文有疑问，欢迎来咱们的 Nebula Graph 论坛交流下心得~~
喜欢这篇文章？来来来，给咱们的 GitHub 点个 star 表鼓励啦~~ 🙇♂️🙇♀️ [手动跪谢]
交流图数据库技术？交个朋友，Nebula Graph 官方小助手微信：NebulaGraphbot 拉你进交流群~~