导入Hive数据Graph

本文以一个示例说明如何使用Exchange将存储在Hive上的数据导入NebulaGraph。

数据集Graph

本文以Graph为例。

在本示例中，该数据集已经存入Hive中名为basketball的数据库中，以player、team、follow和serve四个表存储了所有点和边的信息。以下为各个表的结构。

scala> spark.sql("describe basketball.player").show
+--------+---------+-------+
|col_name|data_type|comment|
+--------+---------+-------+
|playerid|   string|   null|
|     age|   bigint|   null|
|    name|   string|   null|
+--------+---------+-------+

scala> spark.sql("describe basketball.team").show
+----------+---------+-------+
|  col_name|data_type|comment|
+----------+---------+-------+
|    teamid|   string|   null|
|      name|   string|   null|
+----------+---------+-------+

scala> spark.sql("describe basketball.follow").show
+----------+---------+-------+
|  col_name|data_type|comment|
+----------+---------+-------+
|src_player|   string|   null|
|dst_player|   string|   null|
|    degree|   bigint|   null|
+----------+---------+-------+

scala> spark.sql("describe basketball.serve").show
+----------+---------+-------+
|  col_name|data_type|comment|
+----------+---------+-------+
|  playerid|   string|   null|
|    teamid|   string|   null|
|start_year|   bigint|   null|
|  end_year|   bigint|   null|
+----------+---------+-------+

说明：Hive的数据类型bigint与NebulaGraph的int对应。

环境配置Graph

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

硬件规格：
- CPU：1.7 GHz Quad-Core Intel Core i7
- 内存：16 GB

Spark：2.4.7，单机版

Hadoop：2.9.2，伪分布式部署

Hive：2.3.7，Hive Metastore 数据库为 MySQL 8.0.22

NebulaGraph：2.6.0。使用Graph。

前提条件Graph

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

已经Graph并获取如下信息：
- Graph服务和Meta服务的的IP地址和端口。
- 拥有NebulaGraph写权限的用户名和密码。

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

已经安装Spark。

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

已经安装并开启Hadoop服务，并已启动Hive Metastore数据库（本示例中为 MySQL）。

操作步骤Graph

步骤 1：在NebulaGraph中创建SchemaGraph

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

确认Schema要素。NebulaGraph中的Schema要素如下表所示。

要素名称属性

Tag player name string, age int

Tag team name string

Edge Type follow degree int

Edge Type serve start_year int, end_year int

在NebulaGraph中创建一个图空间basketballplayer，并创建一个Schema，如下所示。

## 创建图空间
nebula> CREATE SPACE basketballplayer \
        (partition_num = 10, \
        replica_factor = 1, \
        vid_type = FIXED_STRING(30));

## 选择图空间basketballplayer
nebula> USE basketballplayer;

## 创建Tag player
nebula> CREATE TAG player(name string, age int);

## 创建Tag team
nebula> CREATE TAG team(name string);

## 创建Edge type follow
nebula> CREATE EDGE follow(degree int);

## 创建Edge type serve
nebula> CREATE EDGE serve(start_year int, end_year int);

更多信息，请参见Graph。

步骤 2：使用Spark SQL确认Hive SQL语句Graph

启动spark-shell环境后，依次运行以下语句，确认Spark能读取Hive中的数据。

scala> sql("select playerid, age, name from basketball.player").show
scala> sql("select teamid, name from basketball.team").show
scala> sql("select src_player, dst_player, degree from basketball.follow").show
scala> sql("select playerid, teamid, start_year, end_year from basketball.serve").show

以下为表basketball.player中读出的结果。

+---------+----+-----------------+
| playerid| age|             name|
+---------+----+-----------------+
|player100|  42|       Tim Duncan|
|player101|  36|      Tony Parker|
|player102|  33|LaMarcus Aldridge|
|player103|  32|         Rudy Gay|
|player104|  32|  Marco Belinelli|
+---------+----+-----------------+
...

步骤 3：修改配置文件Graph

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

{
  # Spark相关配置
  spark: {
    app: {
      name: Nebula Exchange 2.6.0
    }
    driver: {
      cores: 1
      maxResultSize: 1G
    }
    cores {
      max: 16
    }
  }

  # 如果Spark和Hive部署在不同集群，才需要配置连接Hive的参数，否则请忽略这些配置。
  #hive: {
  #  waredir: "hdfs://NAMENODE_IP:9000/apps/svr/hive-xxx/warehouse/"
  #  connectionURL: "jdbc:mysql://your_ip:3306/hive_spark?characterEncoding=UTF-8"
  #  connectionDriverName: "com.mysql.jdbc.Driver"
  #  connectionUserName: "user"
  #  connectionPassword: "password"
  #}

  # NebulaGraph相关配置
  nebula: {
    address:{
      # 以下为NebulaGraph的Graph服务和所有Meta服务所在机器的IP地址及端口。
      # 如果有多个地址，格式为 "ip1:port","ip2:port","ip3:port"。
      # 不同地址之间以英文逗号 (,) 隔开。
      graph:["127.0.0.1:9669"]
      meta:["127.0.0.1:9559"]
    }
    # 填写的账号必须拥有NebulaGraph相应图空间的写数据权限。
    user: root
    pswd: nebula
    # 填写NebulaGraph中需要写入数据的图空间名称。
    space: basketballplayer
    connection {
      timeout: 3000
      retry: 3
    }
    execution {
      retry: 3
    }
    error: {
      max: 32
      output: /tmp/errors
    }
    rate: {
      limit: 1024
      timeout: 1000
    }
  }
  # 处理点
  tags: [
    # 设置Tag player相关信息。
    {
      # NebulaGraph中对应的Tag名称。
      name: player
      type: {
        # 指定数据源文件格式，设置为hive。
        source: hive
        # 指定如何将点数据导入NebulaGraph：Client或SST。
        sink: client
      }

      # 设置读取数据库basketball中player表数据的SQL语句
      exec: "select playerid, age, name from basketball.player"

      # 在fields里指定player表中的列名称，其对应的value会作为NebulaGraph中指定属性。
      # fields和nebula.fields里的配置必须一一对应。
      # 如果需要指定多个列名称，用英文逗号（,）隔开。
      fields: [age,name]
      nebula.fields: [age,name]

      # 指定表中某一列数据为NebulaGraph中点VID的来源。
      vertex:{
        field:playerid
      }

      # 单批次写入 NebulaGraph 的最大数据条数。
      batch: 256

      # Spark 分区数量
      partition: 32
    }
    # 设置Tag team相关信息。
    {
      name: team
      type: {
        source: hive
        sink: client
      }
      exec: "select teamid, name from basketball.team"
      fields: [name]
      nebula.fields: [name]
      vertex: {
        field: teamid
      }
      batch: 256
      partition: 32
    }

  ]

  # 处理边数据
  edges: [
    # 设置Edge type follow相关信息
    {
      # NebulaGraph中对应的Edge type名称。
      name: follow

      type: {
        # 指定数据源文件格式，设置为hive。
        source: hive

        # 指定边数据导入NebulaGraph的方式，
        # 指定如何将点数据导入NebulaGraph：Client或SST。
        sink: client
      }

      # 设置读取数据库basketball中follow表数据的SQL语句。
      exec: "select src_player, dst_player, degree from basketball.follow"

      # 在fields里指定follow表中的列名称，其对应的value会作为NebulaGraph中指定属性。
      # fields和nebula.fields里的配置必须一一对应。
      # 如果需要指定多个列名称，用英文逗号（,）隔开。
      fields: [degree]
      nebula.fields: [degree]

      # 在source里，将follow表中某一列作为边的起始点数据源。
      # 在target里，将follow表中某一列作为边的目的点数据源。
      source: {
        field: src_player
      }

      target: {
        field: dst_player
      }

      # 指定一个列作为 rank 的源（可选）。
      #ranking: rank

      # 单批次写入 NebulaGraph 的最大数据条数。
      batch: 256

      # Spark 分区数量
      partition: 32
    }

    # 设置Edge type serve相关信息
    {
      name: serve
      type: {
        source: hive
        sink: client
      }
      exec: "select playerid, teamid, start_year, end_year from basketball.serve"
      fields: [start_year,end_year]
      nebula.fields: [start_year,end_year]
      source: {
        field: playerid
      }
      target: {
        field: teamid
      }

      # 指定一个列作为 rank 的源（可选）。
      #ranking: rank

      batch: 256
      partition: 32
    }
  ]
}

步骤 4：向NebulaGraph导入数据Graph

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

${SPARK_HOME}/bin/spark-submit --master "local" --class com.vesoft.nebula.exchange.Exchange <nebula-exchange-2.6.0.jar_path> -c <hive_application.conf_path> -h

Note

JAR包有两种获取方式：Graph或者从maven仓库下载。

示例：

${SPARK_HOME}/bin/spark-submit  --master "local" --class com.vesoft.nebula.exchange.Exchange  /root/nebula-exchange/nebula-exchange/target/nebula-exchange-2.6.0.jar  -c /root/nebula-exchange/nebula-exchange/target/classes/hive_application.conf -h

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

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

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

GO FROM "player100" OVER follow;

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

步骤 6：（如有）在NebulaGraph中重建索引Graph

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

最后更新: January 18, 2022

要素	名称	属性
Tag	`player`	`name string, age int`
Tag	`team`	`name string`
Edge Type	`follow`	`degree int`
Edge Type	`serve`	`start_year int, end_year int`