Writing to NebulaGraph Database Developer Edition¶
The Spark connector allows you to write data from a Spark DataFrame to nodes and edges in NebulaGraph Database Developer Edition. The examples below use writer options and configuration objects.
Writing nodes¶
The following samples write data from a Spark DataFrame to nodes of node type Movie in the graph movie.
By using writer options¶
Replace <current-password> with the current password for the specified account. If the password has changed, use the new 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()
By using configuration objects¶
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()
Writing edges¶
The following samples write data from a Spark DataFrame to edges of edge type Act in the graph movie.
By using writer options¶
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()
By using configuration objects¶
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()