导入 CSV 文件数据

本文以一个示例说明如何使用 Exchange 将存储在 HDFS 上的 CSV 文件数据导入 Nebula Graph。

如果您要向 Nebula Graph 导入本地 CSV 文件，参考 CSV 文件导入示例。

数据集

本文以美国 Stanford Network Analysis Platform (SNAP) 提供的 Social Network: MOOC User Action Dataset 以及由公开网络上获取的不重复的 97 个课程名称作为示例数据集，包括：

  • 两类点（user 和 course），共计 7,144 个点。
  • 一种关系（action），共计 411,749 条边。

详细的数据集，您可以从 nebula-web-docker 仓库中下载。

环境配置

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

  • 硬件规格：

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

  • Spark：2.3.0，单机版

  • Hadoop：2.9.2，伪分布式部署

  • Nebula Graph：V1.2.1，使用 Docker Compose 部署。详细信息，参考 使用 Docker Compose 部署 Nebula Graph

前提条件

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

  • 已经完成 Exchange 编译。详细信息，参考 编译 Exchange。本示例中使用 Exchange v1.2.1。

  • 已经安装 Spark。

  • 已经安装并开启 Hadoop 服务。

  • 已经部署并启动 Nebula Graph，并获取：

    • Graph 服务、Meta 服务所在机器的 IP 地址和端口信息。
    • Nebula Graph 数据库的拥有写权限的用户名及其密码。

  • 在 Nebula Graph 中创建图数据模式（Schema）所需的所有信息，包括标签和边类型的名称、属性等。

操作步骤

步骤 1. 在 Nebula Graph 中创建 Schema

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

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

    要素名称属性
    标签（Tag）useruserId int
    标签（Tag）coursecourseId int, courseName string
    边类型（Edge Type）actionactionId int, duration double, label bool, feature0 double, feature1 double, feature2 double, feature3 double

  2. 在 Nebula Graph 里创建一个图空间 csv，并创建一个 Schema，如下所示。

    1. -- 创建图空间
    2. CREATE SPACE csv(partition_num=10, replica_factor=1);
    4. -- 选择图空间 csv
    5. USE csv;
    7. -- 创建标签 user
    8. CREATE TAG user(userId int);
    10. -- 创建标签 course
    11. CREATE TAG course(courseId int, courseName string);
    13. -- 创建边类型 action
    14. CREATE EDGE action (actionId int, duration double, label bool, feature0 double, feature1 double, feature2 double, feature3 double);

关于 Nebula Graph 构图的更多信息，参考《Nebula Graph Database 手册》的 快速开始。

步骤 2. 处理 CSV 文件

确认以下信息：

  1. CSV 文件已经根据 Schema 作了处理。详细操作请参考 Nebula Graph Studio 快速开始。

    说明：Exchange 支持上传有表头或者无表头的 CSV 文件。

  2. CSV 文件必须存储在 HDFS 中，并已获取文件存储路径。

步骤 3. 修改配置文件

完成 Exchange 编译后，进入 nebula-java/tools/exchange 目录，根据 target/classes/application.conf 文件修改 CSV 数据源相关的配置文件。在本示例中，文件被重命名为 csv_application.conf。以下仅详细说明点和边数据的配置信息，本次示例中未使用的配置项已被注释，但是提供了配置说明。Spark 和 Nebula Graph 相关配置，参考 Spark 参数和 Nebula Graph 参数。

  1. {
  2.   # Spark 相关配置
  3.   spark: {
  4.     app: {
  5.       name: Spark Writer
  6.     }
  7.     driver: {
  8.       cores: 1
  9.       maxResultSize: 1G
  10.     }
  11.     cores {
  12.       max: 16
  13.     }
  14.   }
  15.   # Nebula Graph 相关配置
  16.   nebula: {
  17.     address:{
  18.       # 以下为 Nebula Graph 的 Graph 服务和 Meta 服务所在机器的 IP 地址及端口
  19.       # 如果有多个地址，格式为 "ip1:port","ip2:port","ip3:port"
  20.       # 不同地址之间以英文逗号 (,) 隔开
  21.       graph:["127.0.0.1:3699"]
  22.       meta:["127.0.0.1:45500"]
  23.     }
  24.     # 填写的账号必须拥有 Nebula Graph 相应图空间的写数据权限
  25.     user: user
  26.     pswd: password
  27.     # 填写 Nebula Graph 中需要写入数据的图空间名称
  28.     space: csv
  29.     connection {
  30.       timeout: 3000
  31.       retry: 3
  32.     }
  33.     execution {
  34.       retry: 3
  35.     }
  36.     error: {
  37.       max: 32
  38.       output: /tmp/errors
  39.     }
  40.     rate: {
  41.       limit: 1024
  42.       timeout: 1000
  43.     }
  44.   }
  45.   # 处理标签
  46.   tags: [
  47.     # 设置标签 course 相关信息
  48.     {
  49.       # Nebula Graph 中对应的标签名称。
  50.       name: course
  51.       type: {
  52.         # 指定数据源文件格式，设置为 csv。
  53.         source: csv
  54.         # 指定点数据导入 Nebula Graph 的方式，
  55.         # 可以设置为：client（以客户端形式导入）和 sst（以 SST 文件格式导入）。
  56.         # 关于 SST 文件导入配置，参考文档：导入 SST 文件（https://
  57.         # docs.nebula-graph.com.cn/nebula-exchange/
  58.         # use-exchange/ex-ug-import-sst/）。
  59.         sink: client
  60.       }
  61.       # CSV 文件所在的 HDFS 路径，String 类型，必须以 hdfs:// 开头。
  62.       path: "hdfs://namenode_ip:port/path/to/course.csv"
  64.       # 如果 CSV 文件里不带表头，则写入 [_c0, _c1, _c2, ... _cn]，
  65.       # 表示 CSV 文件中的数据列名，作为 course 各属性值来源。
  66.       # 如果 CSV 文件里有表头，则写入各列名。
  67.       # fields 与 nebula.fields 的顺序必须一一对应。
  68.       fields: [_c0, _c1]
  70.       # 设置 Nebula Graph 中与 CSV 文件各列对应的属性名称，
  71.       # fields 与 nebula.fields 的顺序必须一一对应。
  72.       nebula.fields: [courseId, courseName]
  74.       # Exchange 1.2.1 添加了 csv.fields 参数：
  75.       # 如果配置了 csv.fields，无论 CSV 文件是否有表头，
  76.       # fields 的配置必须与 csv.fields 的配置保持一致。
  77.       # csv.fields: [courseId, courseName]
  79.       # 指定 CSV 中的某一列数据为 Nebula Graph 中点 VID 的来源。
  80.       # vertex.field 的值必须与上述 fields 或者 csv.fields 中的列名保持一致。
  81.       # 如果数据不是 int 类型，则添加 vertex.policy 指定 VID 映射策略，建议设置为 "hash"。
  82.       vertex: {
  83.         field: _c1,
  84.         policy: "hash"
  85.       }
  87.       # 标明数据源中数据分隔方式，默认为英文逗号（,）。
  88.       separator: ","
  90.       # 如果 CSV 文件中有表头，header 设置为 true。
  91.       # 如果 CSV 文件中没有表头，header 设置为 false（默认值）。
  92.       header: false
  94.       # 单次写入 Nebula Graph 的最大点数据量。
  95.       batch: 256
  97.       # Spark 分区数量
  98.       partition: 32
  100.       # isImplicit 的设置说明参考：https://github.com/vesoft-inc/nebula-java/
  101.       # blob/v1.0/tools/exchange/src/main/resources/application.conf
  102.       isImplicit: true
  103.     }
  105.     # 设置标签 user 相关信息
  106.     {
  107.       name: user
  108.       type: {
  109.         source: csv
  110.         sink: client
  111.       }
  112.       path: "hdfs://namenode_ip:port/path/to/user.csv"
  114.       # Exchange 1.2.1 添加了 csv.fields 参数
  115.       # 如果 CSV 文件里不带表头，但是配置了 csv.fields，
  116.       # 则 fields 的配置必须与 csv.fields 保持一致，
  117.       # Exchange 会将 csv.fields 里的设置作为表头。
  118.       # fields 与 nebula.fields 的顺序必须一一对应。
  119.       fields: [userId]
  121.       # 设置 Nebula Graph 中与 CSV 文件各列对应的属性名称，
  122.       # 必须与 fields 或者 csv.fields 的顺序一一对应。
  123.       nebula.fields: [userId]
  125.       # 如果配置了 csv.fields，无论 CSV 文件是否有表头，
  126.       # 均以这个参数指定的名称作为表头，
  127.       # fields 的配置必须与 csv.fields 的配置保持一致，
  128.       # 同时，vertex 的设置必须与 csv.fields 的设置相同。
  129.       csv.fields: [userId]
  131.       # vertex 的值必须与 fields 或者 csv.fields 中相应的列名保持一致。
  132.       vertex: userId
  133.       separator: ","
  134.       header: false
  135.       batch: 256
  136.       partition: 32
  138.       # isImplicit 设置说明，详见 https://github.com/vesoft-inc/nebula-java/blob
  139.       # /v1.0/tools/exchange/src/main/resources/application.conf
  140.       isImplicit: true
  141.     }
  142.   ]
  143.   # 处理边数据
  144.   edges: [
  145.     # 设置边类型 action 相关信息
  146.     {
  147.       # Nebula Graph 中对应的边类型名称。
  148.       name: action
  149.       type: {
  150.         # 指定数据源文件格式，设置为 csv。
  151.         source: csv
  153.         # 指定边数据导入 Nebula Graph 的方式，
  154.         # 可以设置为：client（以客户端形式导入）和 sst（以 SST 文件格式导入）。
  155.         # 关于 SST 文件导入配置，参考文档：导入 SST 文件（https://
  156.         # docs.nebula-graph.com.cn/nebula-exchange/
  157.         # use-exchange/ex-ug-import-sst/）。
  158.         sink: client
  159.       }
  161.       # 指定 CSV 文件所在的 HDFS 路径，String 类型，必须以 hdfs:// 开头。
  162.       path: "hdfs://namenode_ip:port/path/to/actions.csv"
  164.       # 如果 CSV 文件里不带表头，
  165.       # 则写入 [_c0, _c1, _c2, ... _cn]，
  166.       # 依次表示 CSV 文件中各数据列，作为 action 各属性值来源。
  167.       # 如果 CSV 文件里有表头，则写入各列名。
  168.       # fields 与 nebula.fields 的顺序必须一一对应。
  169.       fields: [_c0, _c3, _c4, _c5, _c6, _c7, _c8]
  171.       # Nebula Graph 中 action 的属性名称，必须与 fields 里的列顺序一一对应。
  172.       nebula.fields: [actionId, duration, feature0, feature1, feature2, feature3, label]
  174.       # Exchange 1.2.1 添加了 csv.fields 参数：
  175.       # 如果配置了 csv.fields，无论 CSV 文件是否有表头，
  176.       # 均以这个参数指定的名称作为表头，
  177.       # fields 的配置必须与 csv.fields 的配置保持一致。
  178.       # csv.fields: [actionId, duration, feature0, feature1, feature2, feature3, label]
  180.       # 边起点和边终点 VID 数据来源，
  181.       # 如果不是 int 类型数据，则添加 policy 指定 VID 映射策略，建议设置为 "hash"。
  182.       source: _c1
  183.       target: {
  184.         field: _c2
  185.         policy: "hash"
  186.       }
  188.       # 标明数据源中数据分隔方式，默认为英文逗号（,）。
  189.       separator: ","
  191.       # 如果 CSV 文件中有表头，header 设置为 true。
  192.       # 如果 CSV 文件中没有表头，header 设置为 false（默认）。
  193.       header: false
  195.       # 单次向 Nebula Graph 写入的最大边数据量。
  196.       batch: 256
  198.       # 设置 Spark 分区数量。
  199.       partition: 32
  200.       isImplicit: true
  201.     }
  202.   ]
  203.   # 如果还有其他边，再添加其他边类型相关的设置。
  204. }

步骤 4. （可选）检查配置文件是否正确

完成配置后，运行以下命令检查配置文件格式是否正确。关于参数的说明，参考 导入命令参数。

  1. $SPARK_HOME/bin/spark-submit --master "local" --class com.vesoft.nebula.tools.importer.Exchange /path/to/exchange-1.2.1.jar -c /path/to/conf/csv_application.conf -D

步骤 5. 向 Nebula Graph 导入数据

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

  1. $SPARK_HOME/bin/spark-submit --master "local" --class com.vesoft.nebula.tools.importer.Exchange /path/to/exchange-1.2.1.jar -c /path/to/conf/csv_application.conf

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

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

  1. GO FROM 1 OVER action;

如果返回边终点（action._dst）即表明数据已导入。

您也可以使用 db_dump 工具统计数据是否已经全部导入。详细的使用信息参考 Dump Tool。

步骤 7. （可选）在 Nebula Graph 中重构索引

导入数据后，您可以在 Nebula Graph 中重新创建并重构索引。详细信息，参考《Nebula Graph Database 手册》。