导入 Neo4j 数据Graph

{
  # Spark 相关配置
  spark: {
    app: {
      name: Spark Writer
    }

    driver: {
      cores: 1
      maxResultSize: 1G
    }

    cores {
      max: 16
    }
  }

  # NebulaGraph 相关配置
  nebula: {
    address:{
      # 以下为 NebulaGraph 的 Graph 服务和 Meta 服务所在机器的 IP 地址及端口。
      # 如果有多个地址，格式为 "ip1:port","ip2:port","ip3:port"。
      # 不同地址之间以英文逗号 (,) 隔开。
      graph:["127.0.0.1:3699"]
      meta:["127.0.0.1:45500"]
    }
    # 填写的账号必须拥有 NebulaGraph 相应图空间的写数据权限。
    user: user
    pswd: password
    # 填写 NebulaGraph 中需要写入数据的图空间名称。
    space: test
    connection {
      timeout: 3000
      retry: 3
    }
    execution {
      retry: 3
    }
    error: {
      max: 32
      output: /tmp/errors
    }
    rate: {
      limit: 64M
      timeout: 1000
    }
  }

  # 处理点数据
  tags: [
    # 设置标签相关信息
    {
    name: tagA
    # 设置 Neo4j 数据库服务器地址，String 类型，格式必须为 "bolt://ip:port"。
    server: "bolt://127.0.0.1:7687"

    # Neo4j 数据库登录账号和密码。
    user: neo4j
    password: neo4j

    # 传输是否加密，默认值为 false，表示不加密；设置为 true 时，表示加密。
    encryption: false

    # 设置源数据所在 Neo4j 数据库的名称。如果您使用 Community Edition Neo4j，不支持这个参数。
    # database: graph.db

    type: {
        # 指定数据来源，设置为 neo4j。
        source: neo4j
        # 指定点数据导入 NebulaGraph 的方式，
        # 可以设置为：client（以客户端形式导入）和 sst（以 SST 文件格式导入）。
        # 关于 SST 文件导入配置，参考文档：导入 SST 文件（https://
        # docs.nebula-graph.com.cn/nebula-exchange/
        # use-exchange/ex-ug-import-sst/）。
        sink: client
    }

    # 指定 NebulaGraph Schema 中标签对应的属性名称，以列表形式列出，
    # 与 tags.fields 列表中的属性名称一一对应，形成映射关系，
    # 多个属性名称之间以英文逗号（,）隔开。
    nebula.fields: [idInt, idString, tdouble, tboolean]

    # 指定源数据中与 NebulaGraph 标签对应的属性名称，
    # 以列表形式列出，多个属性名称之间以英文逗号（,）隔开，
    # 列出的属性名称必须与 exec 中列出的属性名称保持一致。
    fields       : [idInt, idString, tdouble, tboolean]

    # 将源数据中某个属性的值作为 NebulaGraph 点 VID 的来源，
    # 如果属性为 int 或者 long 类型，使用 vertex 设置 VID 列。
    vertex: idInt
    # 如果数据不是 int 类型，则添加 vertex.policy 指定 VID 映射策略，建议设置为 "hash"。
    # vertex {
    #     field: name
    #     policy: "hash"
    #}

    # Spark 的分区数量，默认值为 32。
    partition: 10

    # 单次写入 NebulaGraph 的点数据量，默认值为 256。
    batch: 2000

    # 设置保存导入进度信息的目录，用于断点续传，
    # 如果未设置，表示不启用断点续传。
    check_point_path: "file:///tmp/test"

    # 写入 Cypher 语句，从 Neo4j 数据库中检索打了某种标签的点的属性，并指定别名
    # Cypher 语句不能以英文分号（`;`）结尾。
    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"
    }
    # 如果有多个标签，则参考以上说明添加更多的标签相关信息。
  ]

  # 处理边数据
  edges: [
    {
      # 指定 NebulaGraph 中的边类型名称。
      name: edgeAB
      type: {
        # 指定数据来源，设置为 neo4j。
        source: neo4j

        # 指定边数据导入 NebulaGraph 的方式，
        # 可以设置为：client（以客户端形式导入）和 sst（以 SST 文件格式导入）。
        # 关于 SST 文件导入配置，参考文档：导入 SST 文件（https://
        # docs.nebula-graph.com.cn/nebula-exchange/
        # use-exchange/ex-ug-import-sst/）。
        sink: client
      }

      # 设置 Neo4j 数据库的信息，包括 IP 地址、端口、用户名和密码。
      server: "bolt://127.0.0.1:7687"
      user: neo4j
      password: neo4j

      # 写入 Cypher 语句，表示从 Neo4j 数据库中查询关系属性。
      # Cypher 语句不能以英文分号（`;`）结尾。
      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)"

      # 在 fields 里指定 Neo4j 中关系的属性名称，其对应的 value
      # 会作为 NebulaGraph 中 edgeAB 的属性（nebula.fields）值来源
      # fields 和 nebula.fields 里的配置必须一一对应
      # 如果需要指定多个列名称，用英文逗号（,）隔开
      fields: [r.idInt, r.idString, r.tdouble, r.tboolean]
      nebula.fields: [idInt, idString, tdboule, tboolean]      

      # 指定源数据中某个属性的值作为 NebulaGraph 边的起始点 VID。
      # 如果属性为 `int` 或者 `long` 类型，使用 source.field 设置起始点 VID 列
      # source.field: field_name
      # 如果不是上述类型的属性，则添加 source.policy 指定 VID 映射策略，建议设置为 "hash"。
      source: {
        field: a.idInt
        policy: "hash"
      }

      # 指定源数据中某个属性的值作为 NebulaGraph 边的终点 VID。
      # 如果属性为 `int` 或者 `long` 类型，使用 target.field 设置终点 VID 列
      # target.field: field_name
      # 如果不是上述类型的属性，则添加 target.policy 指定 VID 映射策略，建议设置为 "hash"。
      target: {
        field: b.idInt
        policy: "hash"
      }

      # 将源数据中某一列数据作为 NebulaGraph 中边的 Rank 值来源。
      ranking: idInt

      # Spark 的分区数量，默认值为 32。
      # 为减轻 Neo4j 的排序压力，将 partition 设置为 1
      partition: 1



      # 单次写入 NebulaGraph 的点数据量，默认值为 256
      batch: 1000

      # 设置保存导入进度信息的目录，用于断点续传。如果未设置，表示不启用断点续传。
      check_point_path: /tmp/test
    }
  ]
}

$SPARK_HOME/bin/spark-submit --master "local[10]" --class com.vesoft.nebula.tools.importer.Exchange target/exchange-1.x.y.jar -c /path/to/conf/neo4j_application.conf -D

$SPARK_HOME/bin/spark-submit --master "local[10]" --class com.vesoft.nebula.tools.importer.Exchange target/exchange-1.x.y.jar -c /path/to/conf/neo4j_application.conf

GO FROM <tagA_VID> OVER edgeAB;