Spark 连接器指南¶

Spark 连接器是用于实现悦数图数据库开发版与 Apache Spark 间数据传输的方法库。本文介绍如何设置 Spark 连接器以实现数据传输。

前提条件¶

已安装 Apache Spark 2.4 或 3.0。
悦数图数据库开发版服务组正在运行中。更多信息，请参见快速开始。

操作步骤¶

将 Spark 连接器导入到您的 Spark 项目中。例如，使用 Maven 导入时，需将以下依赖项添加到pom.xml文件：

对于 Spark 2.4：

  <dependency>
    <groupId>com.vesoft</groupId>
    <artifactId>nebula-connector_spark2.4</artifactId>
    <version>5.3.0</version>
  </dependency>

对于 Spark 3.0：

  <dependency>
    <groupId>com.vesoft</groupId>
    <artifactId>nebula-connector_spark3</artifactId>
    <version>5.3.0</version>
  </dependency>

将 Spark DataFrame 中的数据写入悦数图数据库开发版作为特定点类型的点，可参考以下代码：

将<current-password>替换为登录账号的当前密码。若已修改密码，请使用修改后的密码。

  val df = spark.read.json("vertex.json")
  df.show()

  val nebulaConnectionConfig = NebulaConnectionConfig
    .builder()
    .withGraphAddress("127.0.0.1:9669")
    .withUser("root")
    .withPasswd("<current-password>")
    .build()
  val nebulaWriteNodeConfig: WriteNebulaNodeConfig = WriteNebulaNodeConfig
    .builder()
    .withGraphName("nba")
    .withNodeType("node_type_player")
    .withWriteMode(WriteMode.INSERT)
    .withBatchSize(10)
    .build()
  df.write.nebula(nebulaConnectionConfig, nebulaWriteNodeConfig).writeVertices()

其中，vertex.json 文件的示例如下：

{"id":12,"name":"Tom","score":20,"gender": true,"rate": 1.2}
{"id":13,"name":"Bob","score":21,"gender": true,"rate": 1.2}
{"id":14,"name":"Jane","score":22,"gender": true,"rate": 1.2}
{"id":15,"name":"Jena","score":23,"gender": true,"rate": 1.2}
{"id":16,"name":"Nic","score":24,"gender": true,"rate": 1.2}
{"id":17,"name":"Mei","score":25,"gender": true,"rate": 1.2}
{"id":18,"name":"HH","score":26,"gender": true,"rate": 1.2}
{"id":19,"name":"Tyler","score":27,"gender": true,"rate": 1.2}
{"id":20,"name":"Ber","score":28,"gender": true,"rate": 1.2}
{"id":21,"name":"Mercy","score":29,"gender": true,"rate": 1.2}

除了使用配置对象，也支持使用 Writer 参数向悦数图数据库开发版写入数据。更多信息，请参见向悦数图数据库开发版写数据。

将 Spark DataFrame 中的数据写入悦数图数据库开发版作为特定边类型的边，可参考以下代码：

  val df = spark.read.json("spark-connector/example/src/main/resources/edge")
  df.show()

  val nebulaWriteEdgeConfig: WriteNebulaEdgeConfig = WriteNebulaEdgeConfig
    .builder()
    .withGraphName("nba")
    .withEdge("edge_type_follow")
    .withSrcPkFields(List("src", "name1"))
    .withDstPkFields(List("dst", "name2"))
    .withSrcPksAsProperty(false)
    .withDstPksAsProperty(false)
    .withWriteMode(WriteMode.INSERTIGNORE)
    .withBatchSize(10)
    .build()

  df.write.nebula(getNebulaConnectionConfig, nebulaWriteEdgeConfig).writeEdges()

其中，edge.json 文件的示例如下：

{"src":12,"dst":13,"followness":34, "likeness": 1.0, "name1": "Tom", "name2": "Bob"}
{"src":12,"dst":14,"followness":33, "likeness": 2.0, "name1": "Tom", "name2": "Jane"}
{"src":12,"dst":15,"followness":32, "likeness": 3.0, "name1": "Tom", "name2": "Jena"}
{"src":12,"dst":16,"followness":31, "likeness": 4.0, "name1": "Tom", "name2": "Nic"}
{"src":12,"dst":17,"followness":30, "likeness": 5.0, "name1": "Tom", "name2": "Mei"}
{"src":12,"dst":18,"followness":29, "likeness": 6.0, "name1": "Tom", "name2": "HH"}
{"src":12,"dst":19,"followness":28, "likeness": 7.0, "name1": "Tom", "name2": "Tyler"}
{"src":12,"dst":20,"followness":27, "likeness": 8.0, "name1": "Tom", "name2": "Ber"}
{"src":13,"dst":21,"followness":26, "likeness": 9.0, "name1": "Bob", "name2": "Mercy"}

读取悦数图数据库开发版中特定点类型的点并将其加载到 Spark DataFrame 中，可参考以下代码：

  val nebulaConnectionConfig = NebulaConnectionConfig
    .builder()
    .withGraphAddress("127.0.0.1:9669")
    .withUser("root")
    .withPasswd("<current-password>")
    .build()

  val nebulaNodeReadConfig: ReadNebulaConfig = ReadNebulaConfig
    .builder()
    .withGraphName("nba")
    .withTypeName("node_type_player")
    .withReturnCols(List("name"))
    .withBatchSize(10)
    .withPartitionNum(10)
    .build()

  val df = spark.read.nebula(nebulaConnectionConfig, nebulaNodeReadConfig).loadNode()
  df.show()

除了使用配置对象，也支持使用 Reader 参数从悦数图数据库开发版读取数据。更多信息，请参见从悦数图数据库开发版读数据。

读取悦数图数据库开发版中特定边类型的边并将其加载到 Spark DataFrame 中，可参考以下代码：

  val nebulaConnectionConfig = NebulaConnectionConfig
    .builder()
    .withGraphAddress("127.0.0.1:9669")
    .withUser("root")
    .withPasswd("<current-password>")
    .build()

  val nebulaReadEdgeConfig: ReadNebulaConfig = ReadNebulaConfig
    .builder()
    .withGraphName("nba")
    .withTypeName("edge_type_follow")
    .withReturnCols(List("likeness"))
    .withBatchSize(1000)
    .withPartitionNum(10)
    .build()

  val df = spark.read.nebula(getNebulaConnectionConfig, nebulaReadEdgeConfig).loadEdge()
  df.show()

Spark 连接器指南¶

前提条件¶

操作步骤¶

更多信息¶