导入 Hive 数据

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

数据集

本文以 basketballplayer 数据集为例。

在本示例中，该数据集已经存入 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对应。

环境配置

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

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

• Spark：2.4.7，单机版

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

• NebulaGraph ：master。

前提条件

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

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

  • Graph 服务和 Meta 服务的的 IP 地址和端口。
  • 拥有 NebulaGraph 写权限的用户名和密码。

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

• 已经安装 Spark。

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

• 已经启动 Hive Metastore 数据库（本示例中为 MySQL）。

操作步骤

步骤 1：在 NebulaGraph 中创建 Schema

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

1. 确认 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

2. 在 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);
   

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

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

启动 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：修改配置文件

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

{
  # Spark 相关配置
  spark: {
    app: {
      name: NebulaGraph Exchange 3.8.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 服务的地址。
      #如果您的 NebulaGraph 在虚拟网络中，如k8s，请配置 Leader Meta的地址。
      meta:["127.0.0.1:9559"]
    }
    # 填写的账号必须拥有 NebulaGraph 相应图空间的写数据权限。
    user: root
    pswd: nebula
    # 是否使用 RSA 加密的密码。
    # enableRSA: true
    # 使用 RSA 加密密码时的密钥。
    # privateKey: ""
    # 填写 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
      # udf:{
      #            separator:"_"
      #            oldColNames:[field-0,field-1,field-2]
      #            newColName:new-field
      #        }
      # 为 VID 增加指定的前缀。例如 VID 为 12345，增加前缀 tag1 后为 tag1_12345。下划线无法修改。
      # prefix:"tag1"
      # 对 string 类型的 VID 进行哈希化操作。
      # policy:hash
      }

      # 过滤规则。符合过滤规则的数据会被导入NebulaGraph。
      # filter: "name='Tom'"

      # 批量操作类型，包括 INSERT、UPDATE 和 DELETE。默认为 INSERT。
      #writeMode: INSERT

      # 批量删除时是否删除该点关联的出边和入边。`writeMode`为`DELETE`时该参数生效。
      #deleteEdge: false

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

      # 数据写入 NebulaGraph 时需要创建的分区数。
      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
      # udf:{
      #            separator:"_"
      #            oldColNames:[field-0,field-1,field-2]
      #            newColName:new-field
      #        }
      # 为 VID 增加指定的前缀。例如 VID 为 12345，增加前缀 tag1 后为 tag1_12345。下划线无法修改。
      # prefix:"tag1"
      # 对 string 类型的 VID 进行哈希化操作。
      # policy:hash
      }

      target: {
        field: dst_player
      # udf:{
      #            separator:"_"
      #            oldColNames:[field-0,field-1,field-2]
      #            newColName:new-field
      #        }
      # 为 VID 增加指定的前缀。例如 VID 为 12345，增加前缀 tag1 后为 tag1_12345。下划线无法修改。
      # prefix:"tag1"
      # 对 string 类型的 VID 进行哈希化操作。
      # policy:hash
      }

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

      # 过滤规则。符合过滤规则的数据会被导入NebulaGraph。
      # filter: "name='Tom'"

      # 批量操作类型，包括 INSERT、UPDATE 和 DELETE。默认为 INSERT。
      #writeMode: INSERT

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

      # 数据写入 NebulaGraph 时需要创建的分区数。
      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 导入数据

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

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

Note

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

示例：

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

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

访问 Kerberos 认证的 HDFS

使用 Kerberos 进行安全认证时，需使用以下两种方式之一访问 Kerberos 认证的 HDFS。

• 在命令中设置 Kerberos 配置文件

  在命令中配置--conf和--files，例如：

  ${SPARK_HOME}/bin/spark-submit --master xxx  --num-executors 2 --executor-cores 2 --executor-memory 1g \
  --conf "spark.driver.extraJavaOptions=-Djava.security.krb5.conf=./krb5.conf" \
  --conf "spark.executor.extraJavaOptions=-Djava.security.krb5.conf=./krb5.conf" \
  --files /local/path/to/xxx.keytab,/local/path/to/krb5.conf \
  --class  com.vesoft.nebula.exchange.Exchange  \
  exchange.jar -c xx.conf
  

  --conf中的文件路径有如下两种配置方式：

  • 配置文件的绝对路径。要求所有 YARN 或者 Spark 机器相同路径下都有对应文件。
  • （YARN 模式下推荐）配置文件的相对路径（例如./krb5.conf）。通过--files上传的资源文件就在 Java 虚拟机或者 JAR 的工作目录下。

  --files中的文件必须存储在执行spark-submit命令的机器上。

• 不使用命令

  将 Spark 和 Kerberos 认证的 Hadoop 部署在相同集群内，共用 HDFS 和 YARN，然后在 Spark 的spark-env.sh中增加配置export HADOOP_HOME=<hadoop_home_path>。

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

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

LOOKUP ON player YIELD id(vertex);

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

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

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

---

最后更新: May 9, 2024