Skip to content
Developer Edition runs on one host and does not require a product license file. Need multi-host deployment or performance testing at scale? Explore Enterprise Edition

Spark connector guide

The Spark connector is a library that enables data transfer between NebulaGraph Database Developer Edition and Apache Spark. This topic introduces how to set up the Spark connector to read and write data from and to NebulaGraph Database Developer Edition.

Prerequisites

Steps

  1. Import the Spark connector into your Spark project. For example, to import with Maven, add the following dependency to the pom.xml file:

    • For Spark 2.4: 
        <dependency>
    <groupId>com.vesoft</groupId>
    <artifactId>nebula-connector_spark2.4</artifactId>
    <version>5.3.0</version>
  </dependency>
    • For Spark 3.0: 
        <dependency>
    <groupId>com.vesoft</groupId>
    <artifactId>nebula-connector_spark3</artifactId>
    <version>5.3.0</version>
  </dependency>

  2. To write data from a Spark DataFrame into NebulaGraph Database Developer Edition as nodes of a specific node type, see the following code snippet:

    Replace <current-password> with the current password for the specified account. If the password has changed, use the new 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()

    Example of the vertex.json file:

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

    In addition to using configuration objects, you can also use writer options to write data to NebulaGraph Database Developer Edition. For more information, see Writing to NebulaGraph Database Developer Edition.

  3. To write data from a Spark DataFrame into NebulaGraph Database Developer Edition as edges of a specific edge type, see the following code snippet:

      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()

    Example of the edge.json file:

    {"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. To read nodes of a specific node type from NebulaGraph Database Developer Edition and load them into a Spark DataFrame, see the following code snippet:

      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()

    In addition to using configuration objects, you can also use reader options to read data from NebulaGraph Database Developer Edition. For more information, see Reading from NebulaGraph Database Developer Edition.

  5. To read edges of a specific edge type from NebulaGraph Database Developer Edition and load them into a Spark DataFrame, see the following code snippet:

      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()

More information