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

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 参数
• 配置类