导入Hive数据

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

数据集

本文以basketballplayer数据集为例。

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

  1. scala> spark.sql("describe basketball.player").show
  2. +--------+---------+-------+
  3. |col_name|data_type|comment|
  4. +--------+---------+-------+
  5. |playerid| string| null|
  6. | age| bigint| null|
  7. | name| string| null|
  8. +--------+---------+-------+
  9. scala> spark.sql("describe basketball.team").show
  10. +----------+---------+-------+
  11. | col_name|data_type|comment|
  12. +----------+---------+-------+
  13. | teamid| string| null|
  14. | name| string| null|
  15. +----------+---------+-------+
  16. scala> spark.sql("describe basketball.follow").show
  17. +----------+---------+-------+
  18. | col_name|data_type|comment|
  19. +----------+---------+-------+
  20. |src_player| string| null|
  21. |dst_player| string| null|
  22. | degree| bigint| null|
  23. +----------+---------+-------+
  24. scala> spark.sql("describe basketball.serve").show
  25. +----------+---------+-------+
  26. | col_name|data_type|comment|
  27. +----------+---------+-------+
  28. | playerid| string| null|
  29. | teamid| string| null|
  30. |start_year| bigint| null|
  31. | end_year| bigint| null|
  32. +----------+---------+-------+

说明:Hive的数据类型bigint与Nebula Graph的int对应。

环境配置

本文示例在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

  • Nebula Graph:2.5.1。使用Docker Compose部署

前提条件

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

  • 已经安装部署Nebula Graph并获取如下信息:

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

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

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

  • 已经安装Spark。

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

  • 已经安装并开启Hadoop服务,并已启动Hive Metastore数据库(本示例中为 MySQL)。

操作步骤

步骤 1:在Nebula Graph中创建Schema

分析数据,按以下步骤在Nebula Graph中创建Schema:

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

    要素名称属性
    Tagplayername string, age int
    Tagteamname string
    Edge Typefollowdegree int
    Edge Typeservestart_year int, end_year int
  2. 在Nebula Graph中创建一个图空间basketballplayer,并创建一个Schema,如下所示。

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

更多信息,请参见快速开始

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

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

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

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

  1. +---------+----+-----------------+
  2. | playerid| age| name|
  3. +---------+----+-----------------+
  4. |player100| 42| Tim Duncan|
  5. |player101| 36| Tony Parker|
  6. |player102| 33|LaMarcus Aldridge|
  7. |player103| 32| Rudy Gay|
  8. |player104| 32| Marco Belinelli|
  9. +---------+----+-----------------+
  10. ...

步骤 3:修改配置文件

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

  1. {
  2. # Spark相关配置
  3. spark: {
  4. app: {
  5. name: Nebula Exchange 2.5.1
  6. }
  7. driver: {
  8. cores: 1
  9. maxResultSize: 1G
  10. }
  11. cores {
  12. max: 16
  13. }
  14. }
  15. # 如果Spark和Hive部署在不同集群,才需要配置连接Hive的参数,否则请忽略这些配置。
  16. #hive: {
  17. # waredir: "hdfs://NAMENODE_IP:9000/apps/svr/hive-xxx/warehouse/"
  18. # connectionURL: "jdbc:mysql://your_ip:3306/hive_spark?characterEncoding=UTF-8"
  19. # connectionDriverName: "com.mysql.jdbc.Driver"
  20. # connectionUserName: "user"
  21. # connectionPassword: "password"
  22. #}
  23. # Nebula Graph相关配置
  24. nebula: {
  25. address:{
  26. # 以下为Nebula Graph的Graph服务和所有Meta服务所在机器的IP地址及端口。
  27. # 如果有多个地址,格式为 "ip1:port","ip2:port","ip3:port"。
  28. # 不同地址之间以英文逗号 (,) 隔开。
  29. graph:["127.0.0.1:9669"]
  30. meta:["127.0.0.1:9559"]
  31. }
  32. # 填写的账号必须拥有Nebula Graph相应图空间的写数据权限。
  33. user: root
  34. pswd: nebula
  35. # 填写Nebula Graph中需要写入数据的图空间名称。
  36. space: basketballplayer
  37. connection {
  38. timeout: 3000
  39. retry: 3
  40. }
  41. execution {
  42. retry: 3
  43. }
  44. error: {
  45. max: 32
  46. output: /tmp/errors
  47. }
  48. rate: {
  49. limit: 1024
  50. timeout: 1000
  51. }
  52. }
  53. # 处理点
  54. tags: [
  55. # 设置Tag player相关信息。
  56. {
  57. # Nebula Graph中对应的Tag名称。
  58. name: player
  59. type: {
  60. # 指定数据源文件格式,设置为hive。
  61. source: hive
  62. # 指定如何将点数据导入Nebula Graph:Client或SST。
  63. sink: client
  64. }
  65. # 设置读取数据库basketball中player表数据的SQL语句
  66. exec: "select playerid, age, name from basketball.player"
  67. # 在fields里指定player表中的列名称,其对应的value会作为Nebula Graph中指定属性。
  68. # fields和nebula.fields里的配置必须一一对应。
  69. # 如果需要指定多个列名称,用英文逗号(,)隔开。
  70. fields: [age,name]
  71. nebula.fields: [age,name]
  72. # 指定表中某一列数据为Nebula Graph中点VID的来源。
  73. vertex:{
  74. field:playerid
  75. }
  76. # 单批次写入 Nebula Graph 的最大数据条数。
  77. batch: 256
  78. # Spark 分区数量
  79. partition: 32
  80. }
  81. # 设置Tag team相关信息。
  82. {
  83. name: team
  84. type: {
  85. source: hive
  86. sink: client
  87. }
  88. exec: "select teamid, name from basketball.team"
  89. fields: [name]
  90. nebula.fields: [name]
  91. vertex: {
  92. field: teamid
  93. }
  94. batch: 256
  95. partition: 32
  96. }
  97. ]
  98. # 处理边数据
  99. edges: [
  100. # 设置Edge type follow相关信息
  101. {
  102. # Nebula Graph中对应的Edge type名称。
  103. name: follow
  104. type: {
  105. # 指定数据源文件格式,设置为hive。
  106. source: hive
  107. # 指定边数据导入Nebula Graph的方式,
  108. # 指定如何将点数据导入Nebula Graph:Client或SST。
  109. sink: client
  110. }
  111. # 设置读取数据库basketball中follow表数据的SQL语句。
  112. exec: "select src_player, dst_player, degree from basketball.follow"
  113. # 在fields里指定follow表中的列名称,其对应的value会作为Nebula Graph中指定属性。
  114. # fields和nebula.fields里的配置必须一一对应。
  115. # 如果需要指定多个列名称,用英文逗号(,)隔开。
  116. fields: [degree]
  117. nebula.fields: [degree]
  118. # 在source里,将follow表中某一列作为边的起始点数据源。
  119. # 在target里,将follow表中某一列作为边的目的点数据源。
  120. source: {
  121. field: src_player
  122. }
  123. target: {
  124. field: dst_player
  125. }
  126. # 单批次写入 Nebula Graph 的最大数据条数。
  127. batch: 256
  128. # Spark 分区数量
  129. partition: 32
  130. }
  131. # 设置Edge type serve相关信息
  132. {
  133. name: serve
  134. type: {
  135. source: hive
  136. sink: client
  137. }
  138. exec: "select playerid, teamid, start_year, end_year from basketball.serve"
  139. fields: [start_year,end_year]
  140. nebula.fields: [start_year,end_year]
  141. source: {
  142. field: playerid
  143. }
  144. target: {
  145. field: teamid
  146. }
  147. batch: 256
  148. partition: 32
  149. }
  150. ]
  151. }

步骤 4:向Nebula Graph导入数据

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

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

Note

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

示例:

  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.1.jar -c /root/nebula-spark-utils/nebula-exchange/target/classes/hive_application.conf -h

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

步骤 5:(可选)验证数据

用户可以在Nebula Graph客户端(例如Nebula Graph Studio)中执行查询语句,确认数据是否已导入。例如:

  1. GO FROM "player100" OVER follow;

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

步骤 6:(如有)在Nebula Graph中重建索引

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


最后更新: August 11, 2021