导入SST文件数据

    Exchange支持两种数据导入模式：

    • 直接将数据源的数据通过nGQL语句的形式导入Nebula Graph。

    • 将数据源的数据生成SST文件，然后借助Console将SST文件导入Nebula Graph。

    下文将介绍生成SST文件并用其导入数据的适用场景、实现方法、前提条件、操作步骤等内容。

    适用场景

    • 适合在线业务，因为生成时几乎不会影响业务（只是读取Schema），导入速度快。

    • 适合数据源数据量较大的场景，导入速度快。

    Nebula Graph底层使用RocksDB作为键值型存储引擎。RocksDB是基于硬盘的存储引擎，提供了一系列API用于创建及导入SST格式的文件，有助于快速导入海量数据。

    SST文件是一个内部包含了任意长度的有序键值对集合的文件，用于高效地存储大量键值型数据。生成SST文件的整个过程主要由Exchange的Reader、sstProcessor和sstWriter完成。整个数据处理过程如下：

    1. Reader从数据源中读取数据。

    2. sstProcessor根据Nebula Graph的Schema信息生成SST文件，然后上传至HDFS。SST文件的格式请参见。

    3. sstWriter打开一个文件并插入数据。生成SST文件时，Key必须按照顺序写入。

    4. 生成SST文件之后，RocksDB通过方法将SST文件导入到Nebula Graph中。例如：

      调用IngestExternalFile()方法时，RocksDB默认会将文件拷贝到数据目录，并且阻塞RocksDB写入操作。如果SST文件中的键范围覆盖了Memtable键的范围，则将Memtable落盘（flush）到硬盘。将SST文件放置在LSM树最优位置后，为文件分配一个全局序列号，并打开写操作。

    数据集

    本文以为例。

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

      • CPU：1.7 GHz Quad-Core Intel Core i7
      • 内存：16 GB

    • Spark：2.4.7 单机版

    • Hadoop：2.9.2 伪分布式部署

    • Nebula Graph：2.5.0。

    前提条件

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

    • 已经并获取如下信息：

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

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

      • Meta服务配置文件中的--ws_storage_http_port和Storage服务配置文件中的--ws_http_port一致。例如都为19779。

      • Graph服务配置文件中的--ws_meta_http_port和Meta服务配置文件中的--ws_http_port一致。例如都为19559。

      • Schema的信息，包括Tag和Edge type的名称、属性等。

    • 已经编译Exchange，或者直接编译完成的.jar文件。本示例中使用Exchange 2.5.0。

    • 已经安装Spark。

    • 已经安装JDK 1.8或以上版本，并配置环境变量JAVA_HOME。

    • 确认Hadoop服务在所有部署Storage服务的机器上运行正常。

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

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

      1. ## 创建图空间
      2. nebula> CREATE SPACE basketballplayer \
      3.         (partition_num = 10, \
      4.         replica_factor = 1, \
      5.         vid_type = FIXED_STRING(30));
      7. ## 选择图空间basketballplayer
      8. nebula> USE basketballplayer;
      10. ## 创建Tag player
      11. nebula> CREATE TAG player(name string, age int);
      13. ## 创建Tag team
      14. nebula> CREATE TAG team(name string);
      16. ## 创建Edge type follow
      17. nebula> CREATE EDGE follow(degree int);
      19. ## 创建Edge type serve
      20. nebula> CREATE EDGE serve(start_year int, end_year int);

    更多信息，请参见快速开始。

    步骤 2：处理CSV文件

    确认以下信息：

    1. 处理CSV文件以满足Schema的要求。

    2. 获取CSV文件存储路径。

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

    1. {
    2.   # Spark相关配置
    3.   spark: {
    4.     app: {
    5.       name: Nebula Exchange 2.0
    6.     }
    8.     master:local
    10.     driver: {
    11.       cores: 1
    12.       maxResultSize: 1G
    13.     }
    15.     executor: {
    16.         memory:1G
    17.     }
    19.     cores:{
    20.       max: 16
    21.     }
    22.   }
    24.   # Nebula Graph相关配置
    25.   nebula: {
    26.     address:{
    27.       graph:["127.0.0.1:9669"]
    28.       meta:["127.0.0.1:9559"]
    29.     }
    30.     user: root
    31.     pswd: nebula
    32.     space: basketballplayer
    34.     # SST文件相关配置
    35.     path:{
    36.         # 本地临时存放生成的SST文件的目录
    37.         local:"/tmp"
    39.         # SST文件在HDFS的存储路径
    40.         remote:"/sst"
    42.         # HDFS的NameNode地址
    43.         hdfs.namenode: "hdfs://*.*.*.*:9000"
    44.     }
    46.     # 客户端连接参数
    47.     connection {
    48.       # socket连接、执行的超时时间，单位：毫秒。
    49.       timeout: 30000
    50.     }
    52.       # 最大失败数，超过后会退出应用程序。
    53.       max: 32
    54.       # 失败的导入作业将记录在输出路径中。
    55.       output: /tmp/errors
    56.     }
    58.     # 使用谷歌的RateLimiter来限制发送到NebulaGraph的请求。
    59.     rate: {
    60.       # RateLimiter的稳定吞吐量。
    61.       limit: 1024
    63.       # 从RateLimiter获取允许的超时时间，单位：毫秒
    64.       timeout: 1000
    65.     }
    66.   }
    69.   # 处理点
    70.   tags: [
    71.     # 设置Tag player相关信息。
    72.     {
    73.       # 指定Nebula Graph中定义的Tag名称。
    74.       name: player
    75.       type: {
    76.         # 指定数据源，使用CSV。
    77.         source: csv
    79.         # 指定如何将点数据导入Nebula Graph：Client或SST。
    80.         sink: sst
    81.       }
    83.       # 指定CSV文件的路径。
    84.       # 文件存储在HDFS上，用双引号括起路径，以hdfs://开头，例如"hdfs://ip:port/xx/xx"。
    85.       path: "hdfs://*.*.*.*:9000/dataset/vertex_player.csv"
    87.       # 如果CSV文件没有表头，使用[_c0, _c1, _c2, ..., _cn]表示其表头，并将列指示为属性值的源。
    88.       # 如果CSV文件有表头，则使用实际的列名。
    89.       fields: [_c1, _c2]
    91.       # 指定Nebula Graph中定义的属性名称。
    92.       # fields与nebula.fields的顺序必须一一对应。
    93.       nebula.fields: [age, name]
    95.       # 指定一个列作为VID的源。
    96.       # vertex的值必须与上述fields或者csv.fields中的列名保持一致。
    97.       # 目前，Nebula Graph 2.5.0仅支持字符串或整数类型的VID。
    98.       vertex: {
    99.         field:_c0
    100.       }
    102.       # 指定的分隔符。默认值为英文逗号（,）。
    103.       separator: ","
    105.       # 如果CSV文件有表头，请将header设置为true。
    106.       # 如果CSV文件没有表头，请将header设置为false。默认值为false。
    107.       header: false
    109.       # 指定单批次写入Nebula Graph的最大点数量。
    110.       batch: 256
    112.       # 指定Spark分片数量。
    113.       partition: 32
    114.     }
    116.     # 设置Tag team相关信息。
    117.     {
    118.       # 指定Nebula Graph中定义的Tag名称。
    119.       name: team
    120.       type: {
    121.         # 指定数据源，使用CSV。
    122.         source: csv
    124.         # 指定如何将点数据导入Nebula Graph：Client或SST。
    125.         sink: sst
    126.       }
    128.       # 指定CSV文件的路径。
    129.       # 文件存储在HDFS上，用双引号括起路径，以hdfs://开头，例如"hdfs://ip:port/xx/xx"。
    130.       path: "hdfs://*.*.*.*:9000/dataset/vertex_team.csv"
    132.       # 如果CSV文件没有表头，使用[_c0, _c1, _c2, ..., _cn]表示其表头，并将列指示为属性值的源。
    133.       # 如果CSV文件有表头，则使用实际的列名。
    134.       fields: [_c1]
    136.       # 指定Nebula Graph中定义的属性名称。
    137.       # fields与nebula.fields的顺序必须一一对应。
    138.       nebula.fields: [name]
    140.       # 指定一个列作为VID的源。
    141.       # vertex的值必须与上述fields或者csv.fields中的列名保持一致。
    142.       # 目前，Nebula Graph 2.5.0仅支持字符串或整数类型的VID。
    143.       vertex: {
    144.         field:_c0
    145.       }
    147.       # 指定的分隔符。默认值为英文逗号（,）。
    148.       separator: ","
    150.       # 如果CSV文件有表头，请将header设置为true。
    151.       # 如果CSV文件没有表头，请将header设置为false。默认值为false。
    152.       header: false
    154.       # 指定单批次写入Nebula Graph的最大点数量。
    155.       batch: 256
    157.       # 指定Spark分片数量。
    158.       partition: 32
    159.     }
    162.     # 如果需要添加更多点，请参考前面的配置进行添加。
    163.   ]
    164.   # 处理边
    165.     # 设置Edge type follow相关信息。
    166.     {
    167.       # 指定Nebula Graph中定义的Edge type名称。
    168.       name: follow
    169.       type: {
    170.         # 指定数据源，使用CSV。
    173.         # 指定如何将点数据导入Nebula Graph：Client或SST。
    174.         sink: sst
    175.       }
    177.       # 指定CSV文件的路径。
    178.       # 文件存储在HDFS上，用双引号括起路径，以hdfs://开头，例如"hdfs://ip:port/xx/xx"。
    179.       path: "hdfs://*.*.*.*:9000/dataset/edge_follow.csv"
    181.       # 如果CSV文件没有表头，使用[_c0, _c1, _c2, ..., _cn]表示其表头，并将列指示为属性值的源。
    182.       # 如果CSV文件有表头，则使用实际的列名。
    183.       fields: [_c2]
    185.       # 指定Nebula Graph中定义的属性名称。
    186.       # fields与nebula.fields的顺序必须一一对应。
    187.       nebula.fields: [degree]
    189.       # 指定一个列作为起始点和目的点的源。
    190.       # vertex的值必须与上述fields或者csv.fields中的列名保持一致。
    191.       # 目前，Nebula Graph 2.5.0仅支持字符串或整数类型的VID。
    192.       source: {
    193.         field: _c0
    194.       }
    195.       target: {
    196.         field: _c1
    197.       }
    199.       # 指定的分隔符。默认值为英文逗号（,）。
    200.       separator: ","
    202.       # 指定一个列作为rank的源(可选)。
    204.       #ranking: rank
    206.       # 如果CSV文件有表头，请将header设置为true。
    207.       # 如果CSV文件没有表头，请将header设置为false。默认值为false。
    208.       header: false
    210.       # 指定单批次写入Nebula Graph的最大边数量。
    211.       batch: 256
    213.       # 指定Spark分片数量。
    214.       partition: 32
    215.     }
    217.     # 设置Edge type serve相关信息。
    218.     {
    219.       # 指定Nebula Graph中定义的Edge type名称。
    220.       name: serve
    221.       type: {
    222.         # 指定数据源，使用CSV。
    223.         source: csv
    225.         # 指定如何将点数据导入Nebula Graph：Client或SST。
    226.         sink: sst
    227.       }
    229.       # 指定CSV文件的路径。
    230.       # 文件存储在HDFS上，用双引号括起路径，以hdfs://开头，例如"hdfs://ip:port/xx/xx"。
    231.       path: "hdfs://*.*.*.*:9000/dataset/edge_serve.csv"
    233.       # 如果CSV文件没有表头，使用[_c0, _c1, _c2, ..., _cn]表示其表头，并将列指示为属性值的源。
    234.       # 如果CSV文件有表头，则使用实际的列名。
    235.       fields: [_c2,_c3]
    237.       # 指定Nebula Graph中定义的属性名称。
    238.       # fields与nebula.fields的顺序必须一一对应。
    239.       nebula.fields: [start_year, end_year]
    241.       # 指定一个列作为起始点和目的点的源。
    242.       # vertex的值必须与上述fields或者csv.fields中的列名保持一致。
    243.       # 目前，Nebula Graph 2.5.0仅支持字符串或整数类型的VID。
    244.       source: {
    245.         field: _c0
    246.       }
    247.       target: {
    248.         field: _c1
    249.       }
    251.       # 指定的分隔符。默认值为英文逗号（,）。
    252.       separator: ","
    254.       # 指定一个列作为rank的源(可选)。
    255.       #ranking: _c5
    257.       # 如果CSV文件有表头，请将header设置为true。
    258.       # 如果CSV文件没有表头，请将header设置为false。默认值为false。
    259.       header: false
    261.       # 指定单批次写入Nebula Graph的最大边数量。
    262.       batch: 256
    264.       # 指定Spark分片数量。
    265.       partition: 32
    266.     }
    268.   ]
    269.   # 如果需要添加更多边，请参考前面的配置进行添加。
    270. }

    步骤 4：生成SST文件

    运行如下命令将CSV源文件生成为SST文件。关于参数的说明，请参见命令参数。

    示例：

    1. ${SPARK_HOME}/bin/spark-submit  --master "local" --class com.vesoft.nebula.exchange.Exchange  /root/nebula-spark-utils/nebula-exchange/target/nebula-exchange-2.5.0.jar  -c /root/nebula-spark-utils/nebula-exchange/target/classes/sst_application.conf

    任务执行完成后，可以在HDFS上的/sst目录（nebula.path.remote参数指定）内查看到生成的SST文件。

    使用客户端工具连接Nebula Graph数据库，按如下操作导入SST文件：

    1. 执行命令选择之前创建的图空间。

      1. nebula> USE basketballplayer;

    2. 执行命令下载SST文件：

      示例：

      1. nebula> DOWNLOAD HDFS "hdfs://*.*.*.*:9000/sst";

    3. 执行命令导入SST文件：

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

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

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

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