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

# 向悦数图数据库开发版写数据

Spark 连接器支持将 Spark DataFrame 中的数据写入悦数图数据库开发版中的点和边。本文演示了如何使用 Writer 参数和配置对象向悦数图数据库开发版写入数据。

## 写入点

以下示例将 Spark DataFrame 中的数据写入图`movie`中点类型为`Movie`的点。

### 通过 Writer 参数写入

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

````
      val df = spark.read.option("header", true).csv("movie.csv")
      df
        .write
        .format("com.vesoft.nebula.spark.connector.NebulaDataSource")
        .mode("Overwrite")
        .option("type", "node")
        .option("operate_type", "write")
        .option("graph_address", "127.0.0.1:9669")
        .option("user","root")
        .option("passwd", "<current-password>")
        .option("graph_name", "movie")
        .option("label", "movie")
        .option("write_mode", "insert_ignore")
        .option("batch_size","10")
        .save()

````

### 通过配置对象写入

````
    val df = spark.read.option("header", true).csv("movie.csv")
    val connectionConfig: NebulaConnectionConfig = NebulaConnectionConfig
      .builder()
      .withGraphAddress("127.0.0.1:9669")
      .withUser("root")
      .withPasswd("<current-password>")
      .build()

    val nebulaWriteNodeConfig: WriteNebulaNodeConfig = WriteNebulaNodeConfig
      .builder()
      .withGraphName("movie")
      //.withSchema("/default_schema")
      //.withZonedDatetimeFormat("%Y-%m-%dT%H:%M:%S %z")
      //.withLocalDatetimeFormat("%Y-%m-%dT%H:%M:%S")
      //.withZonedTimeFormat("%H:%M:%S %z")
      //.withLocalTimeFormat("%H:%M:%S")
      .withNodeType("Movie")
      .withWriteMode(WriteMode.INSERTIGNORE)
      .withBatchSize(10)
      .build()
    df.write.nebula(getNebulaConnectionConfig, nebulaWriteNodeConfig).writeNodes()

````

## 写入边

以下示例将 Spark DataFrame 中的数据写入图`movie`中边类型为`Act`的边。

### 通过 Writer 参数写入

````
      val df = spark.read.option("header", true).csv("act.csv")
      df
        .write
        .format("com.vesoft.nebula.spark.connector.NebulaDataSource")
        .mode("Overwrite")
        .option("type","edge")
        .option("operate_type", "write")
        .option("graph_address", "127.0.0.1:9669")
        .option("user", "root")
        .option("passwd", "<current-password>")
        .option("graph_name", "movie")
        .option("label", "Act")
        .option("src_pk_field", "src_col1")
        .option("dst_pk_field", "dst_col1")
        .option("write_mode", "insert_ignore")
        .option("batch_size", "10")
        .save()

````

### 通过配置对象写入

````
    val df = spark.read.option("header", true).csv("act.csv")
    val connectionConfig: NebulaConnectionConfig = NebulaConnectionConfig
      .builder()
      .withGraphAddress("127.0.0.1:9669")
      .withUser("root")
      .withPasswd("<current-password>")
      .build()
    val nebulaWriteEdgeConfig: WriteNebulaEdgeConfig = WriteNebulaEdgeConfig
      .builder()
      .withGraphName("movie")
      //.withSchema("/default_schema")
      //.withZonedDatetimeFormat("%Y-%m-%dT%H:%M:%S %z")
      //.withLocalDatetimeFormat("%Y-%m-%dT%H:%M:%S")
      //.withZonedTimeFormat("%H:%M:%S %z")
      //.withLocalTimeFormat("%H:%M:%S")
      .withEdge("Act")
      .withSrcPkFields(List("actor_id"))
      .withDstPkFields(List("movie_id"))
      .withSrcPksAsProperty(false)
      .withDstPksAsProperty(false)
      .withWriteMode(WriteMode.INSERTIGNORE)
      .withBatchSize(10)
      .build()
    df.write.nebula(getNebulaConnectionConfig, nebulaWriteEdgeConfig).writeEdges()

````

## 更多信息

- [Writer 参数](https://dev-docs.yueshu.com.cn/v5.3.2/clients/spark-connector/config/writer-options/)
- [配置类](https://dev-docs.yueshu.com.cn/v5.3.2/clients/spark-connector/config/spark-options/)