Source: https://dev-docs.yueshu.com.cn/v5.3.2/clients/spark-connector/
Documentation version: v5.3.2

# Spark 连接器指南

[Spark 连接器](https://github.com/vesoft-inc/nebula-spark-connector/tree/v5.3.0)是用于实现悦数图数据库开发版与 Apache Spark 间数据传输的方法库。本文介绍如何设置 Spark 连接器以实现数据传输。

## 前提条件

- 已安装 Apache Spark 2.4 或 3.0。
- 悦数图数据库开发版服务组正在运行中。更多信息，请参见[快速开始](https://dev-docs.yueshu.com.cn/v5.3.2/get-started/deploy/)。

## 操作步骤

1. 

将 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>

````

2. 

将 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 参数向悦数图数据库开发版写入数据。更多信息，请参见[向悦数图数据库开发版写数据](https://dev-docs.yueshu.com.cn/v5.3.2/clients/spark-connector/write/)。

3. 

将 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"}

````

4. 

读取悦数图数据库开发版中特定点类型的点并将其加载到 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 参数从悦数图数据库开发版读取数据。更多信息，请参见[从悦数图数据库开发版读数据](https://dev-docs.yueshu.com.cn/v5.3.2/clients/spark-connector/read/)。

5. 

读取悦数图数据库开发版中特定边类型的边并将其加载到 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 连接器 GitHub 仓库](https://github.com/vesoft-inc/nebula-spark-connector/tree/v5.3.0)
- [从悦数图数据库开发版读数据](https://dev-docs.yueshu.com.cn/v5.3.2/clients/spark-connector/read/)
- [向悦数图数据库开发版写数据](https://dev-docs.yueshu.com.cn/v5.3.2/clients/spark-connector/write/)
- [数据导入配置](https://dev-docs.yueshu.com.cn/v5.3.2/clients/spark-connector/config/)