导入Neo4j数据

    Exchange使用Neo4j Driver 4.0.1实现对Neo4j数据的读取。执行批量导出之前，用户需要在配置文件中写入针对标签（label）和关系类型（Relationship Type）自动执行的Cypher语句，以及Spark分区数，提高数据导出性能。

    Exchange读取Neo4j数据时需要完成以下工作：

    1. Exchange中的Reader会将配置文件中部分的CypherRETURN语句后面的语句替换为COUNT(*)，并执行这个语句，从而获取数据总量，再根据Spark分区数量计算每个分区的起始偏移量和大小。

    2. （可选）如果用户配置了check_point_path目录，Reader会读取目录中的文件。如果处于续传状态，Reader会计算每个Spark分区应该有的偏移量和大小。

    3. 在每个Spark分区里，Exchange中的Reader会在Cypher语句后面添加不同的SKIP和LIMIT语句，调用Neo4j Driver并行执行，将数据分布到不同的Spark分区中。

    4. Reader最后将返回的数据处理成DataFrame。

    至此，Exchange即完成了对Neo4j数据的导出。之后，数据被并行写入Nebula Graph数据库中。

    整个过程如下图所示。

    本文以basketballplayer数据集为例。

    本文示例在MacOS下完成，以下是相关的环境配置信息：

    • 硬件规格：

      • CPU：Intel(R) Xeon(R) CPU E5-2697 v3 @ 2.60GHz

      • CPU 内核数：14

    • Spark：单机版，2.4.6 pre-build for Hadoop 2.7

    • Neo4j：3.5.20 Community Edition

    • Nebula Graph：2.6.0。使用。

    开始导入数据之前，用户需要确认以下信息：

    • 已经安装部署Nebula Graph并获取如下信息：

      • Graph服务和Meta服务的的IP地址和端口。

      • 拥有Nebula Graph写权限的用户名和密码。

    • 已经编译Exchange。详情请参见。本示例中使用Exchange 2.6.0。

    • 已经安装Spark。

    • 了解Nebula Graph中创建Schema的信息，包括Tag和Edge type的名称、属性等。

    步骤 1：在Nebula Graph中创建Schema

    分析数据，按以下步骤在Nebula Graph中创建Schema：

    1. 确认Schema要素。Nebula Graph中的Schema要素如下表所示。

    更多信息，请参见。

    步骤 3：修改配置文件

    编译Exchange后，复制target/classes/application.conf文件设置数据源相关的配置。在本示例中，复制的文件名为neo4j_application.conf。各个配置项的详细说明请参见。

    exec配置说明

    在配置tags.exec或者edges.exec参数时，需要填写Cypher查询语句。为了保证每次查询结果排序一致，并且为了防止在导入时丢失数据，强烈建议在Cypher查询语句中加入ORDER BY子句，同时，为了提高数据导入效率，最好选取有索引的属性作为排序的属性。如果没有索引，用户也可以观察默认的排序，选择合适的属性用于排序，以提高效率。如果默认的排序找不到规律，用户可以根据点或关系的ID进行排序，并且将partition设置为一个尽量小的值，减轻Neo4j的排序压力。

    另外，Exchange需要在不同Spark分区执行不同SKIP和LIMIT的Cypher语句，所以在tags.exec和edges.exec对应的Cypher语句中不能含有SKIP和LIMIT子句。

    tags.vertex或edges.vertex配置说明

    Nebula Graph在创建点和边时会将ID作为唯一主键，如果主键已存在则会覆盖该主键中的数据。所以，假如将某个Neo4j属性值作为Nebula Graph的ID，而这个属性值在Neo4j中是有重复的，就会导致重复ID，它们对应的数据有且只有一条会存入Nebula Graph中，其它的则会被覆盖掉。由于数据导入过程是并发地往Nebula Graph中写数据，最终保存的数据并不能保证是Neo4j中最新的数据。

    check_point_path配置说明

    如果启用了断点续传功能，为避免数据丢失，在断点和续传之间，数据库不应该改变状态，例如不能添加数据或删除数据，同时，不能更改partition数量配置。

    运行如下命令将文件数据导入到Nebula Graph中。关于参数的说明，请参见。

    Note

    JAR包有两种获取方式：自行编译或者从maven仓库下载。

    示例：

    用户可以在返回信息中搜索batchSuccess.<tag_name/edge_name>，确认成功的数量。例如batchSuccess.follow: 300。

    步骤 5：（可选）验证数据

    用户可以在Nebula Graph客户端（例如Nebula Graph Studio）中执行查询语句，确认数据是否已导入。例如：

    用户也可以使用命令SHOW STATS查看统计数据。

    导入数据后，用户可以在Nebula Graph中重新创建并重建索引。详情请参见。