# Dag Controller Introduction

Dag Controller is a component of NebulaGraph Explorer, which was released after several tests and tunings. It mainly solves the problem of integration of OLTP and OLAP, and the problem of graph algorithms in complex scenarios. To learn more, see [https://docs.nebula-graph.io/master/nebula-explorer/deploy-connect/ex-ug-deploy/#dag_controller](https://docs.nebula-graph.io/master/nebula-explorer/deploy-connect/ex-ug-deploy/#dag_controller).

Below is some information I would like to share about Dag Controller. Feel free to leave a comment and let's discuss it together.

I'm sure you're no stranger to OLTP and OLAP, but let me briefly introduce them again here: OLTP is a fast-response, real-time online data processing method. OLAP is an offline, data algorithm method for complex scenarios. For NebulaGraph, OLTP has a variety of query statements, such as GO, MATCH, etc. OLAP has a variety of graph algorithms, such as Pagerank, Louvain, WCC (Weakly Connected Components), K-Core, and Jaccard.

OLTP and OLAP do not exist independently. For example, we can use the subgraph returned by a MATCH statement as input to the PageRank algorithm, and the result of the PageRank can be written to NebulaGraph to continue executing a MATCH statement or other algorithms. Like building blocks, various [OLTP and OLAP](https://www.nebula-graph.io/posts/OLTP-vs-OLAP-what-are-the-differences) can be assembled to form a data processing method for complex scenarios.

The Dag Controller is a system that handles scenarios where OLTP and OLAP need to be linked and executed. 

![oltp-vs-olap](https://www-cdn.nebula-graph.io/nebula-website-5.0/images/blogs/olap-and-oltp/1.png)

## Architecture

![dag-architecture](https://www-cdn.nebula-graph.io/nebula-website-5.0/images/blogs/olap-and-oltp/2.png)

The responsibilities of Dag Controller:

1. It provides an HTTP interface for external communication and access.

It is used for job submission, stopping, deleting and other operations, as well as the configuration of the system environment.

2. Execution of DAG.

When Dag Controller executes DAG, the OLAP will call NebulaGraph Analytics, and the OLTP will call graphd to complete the execution of nGQL.

NebulaGraph Analytics is our graph computing system, supporting Pagerank, Louvain, WCC, Jaccard, and other graph algorithms, supporting HDFS and NebulaGraph data sources.

Graphd, metad, and storaged are components in NebulaGraph. Graphd is mainly responsible for nGQL parsing, storaged is responsible for data storage, and metad is responsible for metadata storage.

# Cases

## Case 1

![dag-use-case](https://www-cdn.nebula-graph.io/nebula-website-5.0/images/blogs/olap-and-oltp/4.png)

The above figure shows a DAG model that runs the pagerank algorithm on a subgraph. First, we get a subgraph using an nGQL statement, and then we run the pagerank algorithm on this subgraph.

This method can be used when our graph size is particularly large and we only want to run the algorithm on part of the graph data.

## Case 2

![olap-and-oltp-use-case](https://www-cdn.nebula-graph.io/nebula-website-5.0/images/blogs/olap-and-oltp/3.png)

The above figure shows a model to compute the shortest paths for two classes of vertices.

First, the vertex IDs of the two classes are obtained separately by using nGQL, and then the vertex IDs of the two classes are passed to the shortestpath algorithm, which computes the paths between the two classes in the full graph.

Each algorithm can be set up to run the algorithm based on the full graph or subgraph.

There are various DAG models, and different DAG models can be built according to different business scenarios.

# Technical implementation

## DAG model

The DAG refers to a directed graph without loops, and an instance of DAG is considered as a job, and a job has multiple task.

The task in Dag Controller can be an nGQL or a graph algorithm such as PageRank, Louvain, SSSP, etc.

When a job is executed, it needs to sort the task first, there is a lot of related code on the internet, so I won't repeat it here.

## Parallel execution

To ensure the execution efficiency of DAG, multiple DAGs need to be executed in parallel. At the same time, within a DAG, the task without upstream and downstream dependencies also needs to be executed in parallel.

How can multiple DAGs and tasks be executed in parallel? In simple terms, the DAGs and tasks are handled by two thread pools respectively.

The details are as follows:

1. When the system starts, allocate the job thread pool and task thread pool to handle the execution of job and task respectively.

2. Unexecuted jobs are periodically obtained from the database and transferred to the job thread pool.

3. When a job is executed, the task is sorted according to its upstream and downstream dependencies, and then it determines whether the execution of all upstreams of each task is completed, and then the task is handed over to the task thread pool for execution when the upstream execution is completed, or waits if the upstream execution is not completed.

4. If the job thread pool is full during job execution, you need to wait to obtain unexecuted jobs at regular intervals. The same is done when the task thread pool is full.

## Type check

Note that the data type verification problem exists between the data input and output of tasks. For example, task2 is the downstream of task1. The input of task2 must be of type int, and the output of task1 must be of type int.

## DAG stop

When you stop a job, you need to stop multiple task that are running in parallel. A task has a preparation phase, a running phase, and in the running phase there are cross-machine, multi-process situations. Stopping a job requires avoiding the orphan process problem.

## Custom algorithm support

We support treating a customer's algorithm as a task for DAG building. First, configure the algorithm-related parameter information in the system. When executing a job, the system is responsible for running the algorithm that corresponds to the task.

# How NebulaGraph Works
1. [NebulaGraph Architecture — A Bird’s Eye View](https://www.nebula-graph.io/posts/nebula-graph-architecture-overview)
2. [Benchmark: NebulaGraph vs Dgraph vs JanusGraph](https://www.nebula-graph.io/posts/benchmarking-mainstraim-graph-databases-dgraph-nebula-graph-janusgraph)
3. [Comparison: Neo4j vs NebulaGraph vs JanusGraph](https://www.nebula-graph.io/posts/performance-comparison-neo4j-janusgraph-nebula-graph)
4. [Deploy the Graph Database on Kubernetes](https://www.nebula-graph.io/posts/how-to-deploy-nebula-graph-in-kubernetes)


Dag controller mainly solves the problem of integration of OLTP and OLAP, and the problem of graph algorithms in complex scenarios. 

Integration  Practice of OLTP and OLAP


# What is geospatial data?

Geospatial data is information related to geospatial entities, such as points, lines, and polygons.

NebulaGraph 2.6 supports geospatial data. You can store, compute, and retrieve geospatial data in NebulaGraph. Geography is a data type supported in NebulaGraph. It is composed of latitude and longitude that represents geospatial data.

## How to use geospatial data in NebulaGraph?

**Create Schema**

The following example shows how to create tags. You can create edge types in the same way.

NebulaGraph currently supports three types of geospatial data: Point, LineString, and Polygon. The following shows how to create geography types and how to insert geospatial data.

```ngql
CREATE TAG any_shape(geo geography);
CREATE TAG only_point(geo geography(point));
CREATE TAG only_linestring(geo geography(linestring));
CREATE TAG only_polygon(geo geography(polygon));
```

When no geography type is specified, it means that you can store data of any type; when a type is specified, it means that only geospatial data of that type can be stored, such as `geography (point)`, which means that you can only store spatial information of points.

**Insert Data**

Insert data in the `geo` column of the `any_shape` tag.

```ngql
INSERT VERTEX any_shape(geo) VALUES "101":(ST_GeogFromText("POINT(120.12 30.16)"));
INSERT VERTEX any_shape(geo) VALUES "102":(ST_GeogFromText("LINESTRING(3 8, 4.7 73.23)"));
INSERT VERTEX any_shape(geo) VALUES "103":(ST_GeogFromText("POLYGON((75.3 45.4, 112.5 53.6, 122.7 25.5, 93.9 28.6, 75.3 45.4))"));
```

Insert data in the `geo` column of the `only_point` tag.

```ngql
INSERT VERTEX only_point(geo) VALUES "201":(ST_Point(120.12，30.16)"));;
```

Insert data in the `geo` column of the `only_linestring` tag.

```ngql
INSERT VERTEX only_linestring(geo) VALUES "302":(ST_GeogFromText("LINESTRING(3 8, 4.7 73.23)"));
```

Insert data in the `geo` column of the `only_polygon` tag.

```ngql
INSERT VERTEX only_polygon(geo) VALUES "403":(ST_GeogFromText("POLYGON((75.3 45.4, 112.5 53.6, 122.7 25.5, 93.9 28.6, 75.3 45.4))"));
```

When the data inserted does not meet the requirements of the specified type, the data insertion fails.

```ngql
(root@nebula) [geo]> INSERT VERTEX only_polygon(geo) VALUES "404":(ST_GeogFromText("POINT((75.3 45.4))"));
[ERROR (-1005)]: Wrong value type: ST_GeogFromText("POINT((75.3 45.4))")
```

We can see that the geospatial data insertion method is rather peculiar, and is very different from the insertion of basic types such as `int`, `string`, and `bool`.

Let's take `ST_GeogFromText("POINT(120.12 30.16)")` as an example, `ST_GeogFromText` is a geographic location information parsing function, which accepts a string type of geographic location data in WKT (Well-Known Text) standard format.

`POINT(120.12 30.16)` represents a geographic point with longitude 120°12′E and latitude 30°16′N; the `ST_GeogFromText` function parses and constructs a geography data object from the WKT parameter, and then the `INSERT` statement stores it in the NebulaGraph in WKB (Well-Known Binary) standard.

## Geospatial functions

The geospatial functions supported by NebulaGraph can be divided into the following main categories:

- Constructing functions

  - `ST_Point(longitude, latitude)`: Constructs a `geography point` object based on a latitude and longitude pair.

- Parsing functions

  - `ST_GeogFromText(wkt_string)`: Parses `geography` objects from the WKT text.

  - `ST_GeogFromWKB(wkb_string)`: Parses `geography` objects from the WKB text. # Not yet supported, because NebulaGraph does not yet support binary strings.

- Format setting functions

  - `ST_AsText(geography)`: Outputs the `geography` object in the WKT text format.

  - `ST_AsBinary(geography)`: Outputs the `geography` object in the WKB text format. # Not yet supported, because NebulaGraph does not yet support binary strings.

- Conversion functions

  - `ST_Centroid(geography)`: Calculates the center of gravity of a `geography` object, which is a `geography point` object.

- The predicate function

  - `ST_Intersects(geography_1, geography_2)`: Determines whether two `geography` objects intersect.

  - `ST_Covers(geography_1, geography_2)`: Determines if the first `geography` object completely covers the second.

  - `ST_CoveredBy(geography_1, geography_2)`: The inverse of ST_Covers.

  - `ST_DWithin(geography_1, geography_2, distance_in_meters)`: Determines if the shortest distance between two `geography` objects is less than the given distance.

- The metric function

  - `ST_Distance(geography_1, geography_2)`: Calculates the distance between two `geography` objects.

These function interfaces follow the OpenGIS Simple Feature Access and ISO SQL/MM standards. For details, see [NebulaGraph doc](https://docs.nebula-graph.io/3.3.0/3.ngql-guide/6.functions-and-expressions/14.geo/).

## Geospatial Index

**What is a geospatial index?**

Geospatial indexes are indexes that can be used to quickly filter data based on the predicate `ST_Intersects` and `ST_Covers` functions. 

NebulaGraph uses the Google S2 library as the geospatial index. 

The S2 library projects the Earth's surface into a tangent square, then recursively quadruples each square surface of the square n times, and uses a space-filling curve, the Hilbert curve, to connect the centers of these small square lattices.

When n is infinitely large, this Hilbert curve almost fills the square.

The S2 library uses a Hilbert curve of order 30.

![](https://www-cdn.nebula-graph.io/nebula-website-5.0/images/blogs/Exploring-Geospatial-Data-with-NebulaGraph/1.jpeg)

The following figure shows that the Earth is filled with Hilbert curves.

![](https://www-cdn.nebula-graph.io/nebula-website-5.0/images/blogs/Exploring-Geospatial-Data-with-NebulaGraph/2.jpeg)


It can be seen that the Earth's surface is divided into cells by these Hilbert curves. For any geographic shape on the earth's surface, such as a city, a river, or a person's location, we can use several of these cells to completely cover the geographic shape.

Each cell is identified by a unique int64 CellID. Thus, the spatial index of a geographic object is the set of S2 cells that are constructed to completely cover the geographic shape.

When constructing an index of a geospatial object, a collection of different S2 cells that completely cover the indexed object is constructed. The indexing query based on spatial predicate functions quickly filters out a large number of irrelevant geographic objects by finding the intersection between the set of S2 cells that cover the queried object and the S2 cells that cover the indexed object.


**Create a geography index**

```ngql
CREATE TAG any_shape_geo_index on any_shape(geo)
```

For geospatial data with the type `point`, it can be represented by an S2 cell of order 30, so a point corresponds to one index entry; for geospatial data with the type `inestring` and `polygon`, we use multiple S2 cells of different levels to cover it, so it will correspond to multiple index entries.

Spatial indexing is used to speed up the lookup of all geo predicates, for example:

```ngql
LOOKUP ON any_shape WHERE ST_Intersects(any_shape.geo, ST_GeogFromText("LINESTRING(3 8, 4.7 73.23)"));
```

When there is no spatial index on the `geo` column of `any_shape`, this statement will first read all the data of `any_shape` into memory and then use it to calculate whether it intersects with the point (3.0, 8.0), which is generally more expensive. When the amount of data in `any_shape` is large, the computation overhead will be unacceptable.

When the `geo` column of `any_shape` has a spatial index, the statement will first use the spatial index to filter out most of the data that intersected by the line, but there will still be some that may be intersected when read into memory, so there is still one more calculation to be done. In this way, the spatial index quickly filters out most of the data that is not likely to intersect at a small cost, and a small percentage is filtered, greatly reducing the computational overhead.

You can store, compute, and retrieve geospatial data in NebulaGraph. Geography is a data type supported in NebulaGraph.

Exploring Geospatial Data with NebulaGraph

Based on the actual demand and subsequent scalability, the NebulaGraph database is used for technical selection. The batch import performance in the actual business scenarios of NebulaGraph needs to be first verified.

This article shares the best practice of using NebulaGraph Spark Connector by performing import jobs via Spark on Yarn distributed tasks and with CSV files on HDFS as the data source.

# About NebulaGraph Spark Connector

The following figure shows the use cases and advantages of NebulaGraph Spark Connector. For more details, see [NebulaGraph Spark Connector](https://docs.nebula-graph.io/3.0.1/nebula-spark-connector/).

<pic>


# Environment Information

- Hardware environment 

| **Name**     |  **Value**  | **Recommend** |
| :----------- | :------: | -------: |
| Local Disk (SSD) |   2 T    | At least 2 T |
| CPU          | 16 C * 4 |    128 C |
| Memory         |  128 GB  |    128 GB |

- Software environment

| **Name**               | **Version**  |
| :--------------------- | :---------: |
| NebulaGraph           |    3.0.0    |
| NebulaGraph Spark Connector |    3.0.0    |
| Hadoop                 | 2.7.2U17-10 |
| Spark                  |   2.4.5U5   |

- Data size

| **Name**     | **Value** |
| :----------- | :----: |
| Data size      | 200 GB  |
| Vertex | 930 million |
| Edge   | 970 million |

# Deployment Plan

- Method: Distributed, 3 nodes
- See [Deploy a NebulaGraph cluster with RPM/DEB package on multiple servers](https://docs.nebula-graph.com.cn/3.0.1/4.deployment-and-installation/2.compile-and-install-nebula-graph/deploy-nebula-graph-cluster/)

There are roughly three steps for the deployment.

1. Download the RPM package of NebulaGraph and install it.
2. Modify configuration files in batch.
3. Start the cluster service.

The above-mentioned three steps are performed with the `root` user. For non-root users, use `sudo` to execute the commands.

## Download and Install the NebulaGraph RPM Package

Execute the following commands:

```bash
wget https://os-cdn.nebula-graph.com.cn/package/3.0.0/nebula-graph-3.0.0.el7.x86_64.rpm
wget https://oss-cdn.nebula-graph.com.cn/package/3.0.0/nebula-graph-3.0.0.el7.x86_64.rpm.sha256sum.txt
rpm -ivh nebula-graph-3.0.0.el7.x86_64.rpm
```

Note: The default installation path is `/usr/local/nebula`. Make sure the memory resources are sufficient.

## Modify Configuration Files in Batch

```bash
sed -i 's?--meta_server_addrs=127.0.0.1:9559?--meta_server_addrs=172.1x.x.15:9559,172.1x.x.176:9559,172.1x.1x.149:9559?g' *.conf
sed -i 's?--local_ip=127.0.0.1?--local_ip=172.1x.1x.149?g' *.conf
sed -i 's?--meta_server_addrs=127.0.0.1:9559?--meta_server_addrs=172.1x.x.15:9559,172.1x.x.176:9559,172.1x.1x.149:9559?g' *.conf
sed -i 's?--local_ip=127.0.0.1?--local_ip=172.1x.x.15?g' *.conf
sed -i 's?--meta_server_addrs=127.0.0.1:9559?--meta_server_addrs=172.1x.x.15:9559,172.1x.x.176:9559,172.1x.1x.149:9559?g' *.conf
sed -i 's?--local_ip=127.0.0.1?--local_ip=172.1x.x.176?g' *.conf
```

Note: The IP addresses in the above commands are internal IP addresses, used for communication between nodes. 


## Start the NebulaGraph Cluster Service

```bash
/usr/local/nebula/scripts/nebula.service start all
```

The above command is to start the cluster service, the following is to check the status of the cluster service.

```bash
ps aux|grep nebula
```

The returned result shows 3 processes for each service.

```bash
/usr/local/nebula/bin/nebula-metad --flagfile /usr/local/nebula/etc/nebula-metad.conf
/usr/local/nebula/bin/nebula-graphd --flagfile /usr/local/nebula/etc/nebula-graphd.conf
/usr/local/nebula/bin/nebula-storaged --flagfile /usr/local/nebula/etc/nebula-storaged.conf
```

Note: If the returned result shows less than 3 processes, run `/usr/local/nebula/scripts/nebula.service start all` several times or run `/usr/local/nebula/scripts/nebula.service restart all` to restart the service.


# Visualization Service

The NebulaGraph Studio is used as the visualize the NebulaGraph cluster service. Visit `http://n01v:7001` to log into the NebulaGraph Studio. (This is my network environment which is not accessible from the Internet.)

<pic>

- Login: 10.x.x.1 (The IP of any node):9669
- Username/Password: `root`/`nebula`

Refer to [Use nGQL (CRUD)](https://docs.nebula-graph.com.cn/3.0.1/2.quick-start/4.nebula-graph-crud) for common-used nGQL statements.


## Use NebulaGraph

Add the Storage nodes to the cluster.

```nebula
ADD HOSTS 172.x.x.121:9779, 172.16.11.218:9779,172.16.12.12:9779;
```

List all nodes and check whether the status of the cluster is `ONLINE by running `SHOW HOSTS` or `SHOW HOSTS META`.
Create a graph space, equivalent to a database in relational databases.
```nebula
CREATE SPACE mylove (partition_num = 15, replica_factor = 3, vid_type = FIXED_STRING(256));//The number of partitions is recommended to be 5 times the number of nodes; the number of replicas is an odd number, generally set to 3; if the data type of vids is string, make the length as long as a reasonable value, otherwise, it takes up too much disk space.
```

Create a Tag, equivalent to an entity.: 

```nebula
CREATE TAG entity (name string NULL, version string NULL);
```  

Create an edge, equivalent to a relationship.

```nebula
CREATE EDGE relation (name string NULL);  
```

Add `limit` when querying data, otherwise, the query may take a long time.

```nebula
match (v) return v limit 100;
```

# Read and Import CSV Files with Spark Connector 

References:

- [NebulaSparkWriterExample (Scala Json) from the official](https://github.com/vesoft-inc/nebula-spark-utils/blob/master/example/src/main/scala/com/vesoft/nebula/examples/connector/NebulaSparkWriterExample.scala)
- [NebulaSparkWriterExample (Java Json) from an ace](https://www.jianshu.com/p/930e0343a28c)

Code example of NebulaSparkWriterExample

```
import com.facebook.thrift.protocol.TCompactProtocol
import com.vesoft.nebula.connector.{
  NebulaConnectionConfig,
  WriteMode,
  WriteNebulaEdgeConfig,
  WriteNebulaVertexConfig
}
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.slf4j.LoggerFactory
object NebulaSparkWriter {
  private val LOG = LoggerFactory.getLogger(this.getClass)
  var ip = ""
  def main(args: Array[String]): Unit = {
    val part = args(0)
    ip = args(1)
    val sparkConf = new SparkConf
    sparkConf
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .registerKryoClasses(Array[Class[_]](classOf[TCompactProtocol]))
    val spark = SparkSession
      .builder()
      .master("local")
      .config(sparkConf)
      .getOrCreate()
    spark.sparkContext.setLogLevel("WARN")
    if("1".equalsIgnoreCase(part)) writeVertex(spark)
    if("2".equalsIgnoreCase(part)) writeEdge(spark)
    spark.close()
  }
  def getNebulaConnectionConfig(): NebulaConnectionConfig = {
    val config =
      NebulaConnectionConfig
        .builder()
        .withMetaAddress(ip + ":9559")
        .withGraphAddress(ip + ":9669")
        .withTimeout(Integer.MAX_VALUE)
        .withConenctionRetry(5)
        .build()
    config
  }
  def writeVertex(spark: SparkSession): Unit = {
    LOG.info("start to write nebula vertices: 1 entity")
    val df = spark.read.option("sep", "\t").csv("/home/2022/project/origin_file/csv/tag/entity/").toDF("id", "name", "version")
    val config = getNebulaConnectionConfig()
    val nebulaWriteVertexConfig: WriteNebulaVertexConfig = WriteNebulaVertexConfig
      .builder()
      .withSpace("mywtt")
      .withTag("entity")
      .withVidField("id")
      .withVidAsProp(false)
      .withUser("root")
      .withPasswd("nebula")
      .withBatch(1800)
      .build()
    df.coalesce(1400).write.nebula(config, nebulaWriteVertexConfig).writeVertices()
  }
  def writeEdge(spark: SparkSession): Unit = {
    LOG.info("start to write nebula edges: 2 entityRel")
    val df = spark.read.option("sep", "\t").csv("/home/2022/project/origin_file/csv/out/rel/relation/").toDF("src", "dst", "name")
    val config = getNebulaConnectionConfig()
    val nebulaWriteEdgeConfig: WriteNebulaEdgeConfig = WriteNebulaEdgeConfig
      .builder()
      .withSpace("mywtt")
      .withEdge("relation")
      .withSrcIdField("src")
      .withDstIdField("dst")
      .withSrcAsProperty(false)
      .withDstAsProperty(false)
      .withUser("root")
      .withPasswd("nebula")
      .withBatch(1800)
      .build()
    df.coalesce(1400).write.nebula(config, nebulaWriteEdgeConfig).writeEdges()
  }
}
```
## Details on NebulaSparkWriterExample 

The following are the details of the functions.

- `spark.sparkContext.setLogLevel("WARN")`: Sets the log print level and prevents the INFO log. 


- `withTimeout(Integer.MAX_VALUE)`: Sets the connection timeout. The timeout can be set as large as possible. The default value is 1 minute. When the number of timeouts is greater than that of retries, the Spark task fails.

- `option("sep", "\t")`: Specifies the separator of CSV files, otherwise, it defaults to 1 column.

- `toDF("src", "dst", "name")`: The schema of the dataset, i.e., converting `Dataset<Row>` to `DataFrame`, otherwise, the `VidField` cannot be specified.

- `withVidField("id")`: This function only works when the column name is set, so a schema must be specified.

- `withVidAsProp(false)`: When the ID is set as the VID by default, data will not be written repeatedly as a property that will not take up disk space.

- `withSrcIdField("src")`: Sets the `IdField` of the source vertex.

- `withDstIdField("dst")`: Sets the `IdField` of the destination vertex.

- `withSrcAsProperty(false)`: Saves disk space.

- `withDstAsProperty(false)`: Saves disk space.

- `withBatch(1000)`: The number of rows of data imported in batch. `WriteMode.UPDATE` is <=512 by default. The value of `WriteMode.INSERT` can be larger (Gigabit NIC/Bandwidth 5Gbps /Local SSD = 1500)

- `coalesce(1500)`: Adjustable according to the number of concurrent tasks. Too much data on a single partition can easily cause executor OOM exceptions.

# Submit Tasks to the Spark Cluster

```spark
nohup spark-submit  --master yarn --deploy-mode client --class com.xxx.nebula.connector.NebulaSparkWriter --conf spark.dynamicAllocation.enabled=false --conf spark.executor.memoryOverhead=10g  --conf spark.blacklist.enabled=false --conf spark.default.parallelism=1000 --driver-memory 10G --executor-memory 12G --executor-cores 4 --num-executors 180 ./example-3.0-SNAPSHOT.jar >  run-csv-nebula.log 2>&1 &
```

## Monitoring with `iotop`

The `iotop` command can be used to monitor the disk IO of the Spark cluster.

```bash
Total DISK READ :      26.61 K/s | Total DISK WRITE :     383.77 M/s
Actual DISK READ:      26.61 K/s | Actual DISK WRITE:     431.75 M/s
```
## Monitoring with `top` 

The `top` command can be used to monitor the CPU and memory of the Spark cluster.

```bash
top - 16:03:01 up 8 days, 28 min,  1 user,  load average: 6.16, 6.53, 4.58
Tasks: 205 total,   1 running, 204 sleeping,   0 stopped,   0 zombie
%Cpu(s): 28.3 us, 14.2 sy,  0.0 ni, 56.0 id,  0.6 wa,  0.0 hi,  0.4 si,  0.5 st
KiB Mem : 13186284+total,  1135004 free, 31321240 used, 99406592 buff/cache
KiB Swap:        0 total,        0 free,        0 used. 99641296 avail Mem 
  PID USER      PR  NI    VIRT    RES    SHR S  %CPU %MEM     TIME+ COMMAND                                                                         
27979 root      20   0 39.071g 0.026t   9936 S 564.6 20.8  83:22.03 nebula-storaged                                                                 
27920 root      20   0 2187476 804036   7672 S 128.2  0.6  17:13.75 nebula-graphd                                                                   
27875 root      20   0 6484644 1.990g   8588 S  58.5  1.6  14:14.22 nebula-metad     
```
## Other Resource Monitoring

<pic>

# Service Optimization

## Configuration Optimization of `nebula-storaged.conf`

Here, I modified the value of parameters in the `nebula-storaged.conf` file.

```bash
# Default reserved bytes for a batch operation.
--rocksdb_batch_size=4096
# Default block cache size used in BlockBasedTable.
# The unit is MB. It is usually set to one third of the memory value. The memory is 128 GB.
--rocksdb_block_cache=44024
############## rocksdb Options ##############
--rocksdb_disable_wal=true
# In JSON, the name and value of each option of rocksdb DBOptions is a string. Such as “option_name”:“option_value”. Use a comma to separate each option.
--rocksdb_db_options={"max_subcompactions":"3","max_background_jobs":"3"}
# In JSON, the name and value of each option of rocksdb ColumnFamilyOptions is a string. Such as “option_name”:“option_value”. Use a comma to separate each option.
--rocksdb_column_family_options={"disable_auto_compactions":"false","write_buffer_size":"67108864","max_write_buffer_number":"4","max_bytes_for_level_base":"268435456"}
# In JSON, the name and value of each option of rocksdb BlockBasedTableOptions is a string. Such as “option_name”:“option_value”. Use a comma to separate each option.
--rocksdb_block_based_table_options={"block_size":"8192"}
# The number of processors per request.
--max_handlers_per_req=10
# The interval time between clusters.
--heartbeat_interval_secs=10
--raft_rpc_timeout_ms=5000
--raft_heartbeat_interval_secs=10
--wal_ttl=14400
# The max value for batch import.
--max_batch_size=1800
# Reduces memory usage when the value is true.
--enable_partitioned_index_filter=true
# The data is indirectly filtered at the bottom storage layer to prevent encountering the super node in the production environment.
--max_edge_returned_per_vertex=10000
```
## Optimizations of the Linux System

```bash
ulimit -c unlimited
ulimit -n 130000
sysctl -w net.ipv4.tcp_slow_start_after_idle=0
sysctl -w net.core.somaxconn=2048
sysctl -w net.ipv4.tcp_max_syn_backlog=2048
sysctl -w net.core.netdev_max_backlog=3000
sysctl -w kernel.core_uses_pid=1
```

# Verify Import Results

```nebula
SUBMIT JOB STATS;
SHOW JOB ${ID}
SHOW STATS;
```
- The entity insertion rate is approximately 27,837 items/s (for this import performance calculation only).

- The relationship insertion rate is about 26,276 items/s (for this import performance calculation only).

- The import performance will be better if the server is better configured. In addition, bandwidth, cross-datacenter or not, disk IO, and even network fluctuations are also factors that affect performance.

# Performance Tests

- Query for the specified entities based on properties.

  ```nebula
  MATCH (v:entity) WHERE v.entity.name == 'Lifespan' RETURN v;
  ```

  It takes 0.002558 (s).

- One-hop queries.

  ```nebula
  MATCH (v1:entity)-[e:propertiesRel]->(v2:attribute) WHERE id(v1) == '70da43c5e46f56c634547c7aded3639aa8a1565975303218e2a92af677a7ee3a' RETURN v2 limit 100;  
  ```
  It takes 0.003571 (s)

- Two-hop queries.

  ```nebula
  MATCH p=(v1:entity)-[e:propertiesRel*1..2]->(v2) WHERE id(v1) == '70da43c5e46f56c634547c7aded3639aa8a1565975303218e2a92af677a7ee3a' RETURN p;
  ```
  It takes 0.005143 (s)

- Query for all the properties of an edge.

  ```nebula
  FETCH PROP ON propertiesRel '70da43c5e46f56c634547c7aded3639aa8a1565975303218e2a92af677a7ee3a' -> '0000002d2e88d7ba6659db83893dedf3b8678f3f80de4ffe3f8683694b63a256' YIELD properties(edge); 
  ```  
  It takes 0.001304 (s)

  ```nebula
  match p=(v:entity{name:"张三"})-[e:entityRel|propertiesRel*1]->(v2) return p;
  ```
  å
  It takes 0.02986 (s)

  ```nebula
  match p=(v:entity{name:"张三"})-[e:entityRel|propertiesRel*2]->(v2) return p; 
  ```

  It takes 0.07937 (s)

  ```nebula
  match p=(v:entity{name:"张三"})-[e:entityRel|propertiesRel*3]->(v2) return p; 
  ```

  It takes 0.269 (s)

  ```nebula
  match p=(v:entity{name:"张三"})-[e:entityRel|propertiesRel*4]->(v2) return p;
  ```

  It takes 3.524859 (s)

  ```nebula 
  match p=(v:entity{name:"张三"})-[e:entityRel|propertiesRel*1..2]->(v2) return p; 
  ```

  It takes 0.072367 (s)

  ```nebula 
  match p=(v:entity{name:"张三"})-[e:entityRel|propertiesRel*1..3]->(v2) return p;
  ```

  It takes 0.279011 (s)

  ```nebula 
  match p=(v:entity{name:"张三"})-[e:entityRel|propertiesRel*1..4]->(v2) return p; 
  It takes 3.728018 (s)
  ```

 - Query for the properties of vertices and the edge in the shortest path in bilateral directions between `A_vid` and `B_vid`.

  ```nebula
  FIND SHORTEST PATH WITH PROP FROM "70da43c5e46f56c634547c7aded3639aa8a1565975303218e2a92af677a7ee3a" TO "0000002d2e88d7ba6659db83893dedf3b8678f3f80de4ffe3f8683694b63a256" OVER * BIDIRECT YIELD path AS p; 
  ```
  It takes 0.003096 (s)

  ```nebula
  FIND ALL PATH FROM "70da43c5e46f56c634547c7aded3639aa8a1565975303218e2a92af677a7ee3a" TO "0000002d2e88d7ba6659db83893dedf3b8678f3f80de4ffe3f8683694b63a256" OVER * WHERE propertiesRel.name is not EMPTY or propertiesRel.name >=0 YIELD path AS p;
  ```

  It takes 0.003656 (s)

# Problems Encountered

## Guava version conflict

```
Caused by: java.lang.NoSuchMethodError: com.google.common.base.Stopwatch.createStarted()Lcom/google/common/base/Stopwatch;
```

After troubleshooting, I found that one of the dependent modules used Guava version 22.0, while the Spark cluster came with 14.0, resulting in a conflict, and it did not work properly. For tasks running on Spark clusters, Spark loads Guava packages with higher priority than its packages.

The package we depend on uses methods that are newer in Guava version 22.0 and not available in version 14.0. Without being able to modify each other's code, the following options are available:

1. The Spark cluster package can be upgraded, which is risky and can cause unknown problems.

2. Rename your Guava package using the Maven plugin.

The second way is used here, using the Maven plugin shade (link: https://maven.apache.org/plugins/maven-shade-plugin/) to rename the package to solve the problem.

```java
<plugin>
    <groupId>org.apache.maven.plugins</groupId>
    <artifactId>maven-shade-plugin</artifactId>
    <version>3.2.4</version>
    <executions>
        <execution>
            <phase>package</phase>
            <goals>
                <goal>shade</goal>
            </goals>
            <configuration>
                <relocations>
                    <relocation>
                        <pattern>com.google.common</pattern>
                        <shadedPattern>my_guava.common</shadedPattern>
                    </relocation>
                </relocations>
                <filters>
                    <filter>
                        <artifact>*:*</artifact>
                        <excludes>
                            <exclude>META-INF/maven/**</exclude>
                            <exclude>META-INF/*.SF</exclude>
                            <exclude>META-INF/*.DSA</exclude>
                            <exclude>META-INF/*.RSA</exclude>
                        </excludes>
                    </filter>
                </filters>
            </configuration>
        </execution>
    </executions>
</plugin>
```
## Spark Blacklist Mechanism
```
Blacklisting behavior can be configured via spark.blacklist.*.
```
The default value of `spark.blacklist.enabled` is `false`. If the value is set to `true`, Spark will no longer schedule tasks to the blacklisted executors. The blacklisting algorithm can be further configured by setting `spark.blacklist`.
  
*This article was written by Enqiu Jia and first published on the WeChat Official Account *只要5分钟不要996*. This article is translated and published with permission.

This is a best practice of using NebulaGraph Spark Connector by performing import jobs via Spark on Yarn distributed tasks and with CSV files on HDFS as the data source.

Offline CSV Import of Billions of Data | Best Practice Using NebulaGraph Spark Connector


![nebula-graph-index-explained-featured](https://user-images.githubusercontent.com/57335825/167098195-bde39149-3b4a-4d52-843d-0b56a2309a97.jpg)

The term Index in NebulaGraph is quite similar to the same term in relational databases, but they are not exactly the same. I noticed that some NebulaGraph users are often confused when getting started with NebulaGraph. Typically, people want to know what exactly NebulaGraph Index is, when should they use it, and how it impacts the performance of NebulaGraph.

Today I'm going to walk you through the Index concept in NebulaGraph and hopefully, this article will answer these questions.

Let's get started!

## What exactly NebulaGraph Index is

To put it short, NebulaGraph Index is only used when the vertextID is not specified, or only when **properties** of vertices or edges are defined in the query conditions. 

Index is only used in a starting entry of a graph query. If a query is in the pattern: (a->b->c, where c is with the condition "foobar"), since the only filter `condition-foobar` is on `c`, this query under the hood will start to look for `c`, and then it walks reversely through the `->` to `b`, and finally to `a`. Thus, the NebulaGraph Index will be used and only be possibly used when locating c.

### Index is used only to seek starting points

We know that in RDBMS, indexing is to create a duplicate of sorted data to enable QUERY with conditional filtering on the sorted data, in order to **accelerate the query in reads** and it also brings additional data writes.

> Note: in RDBMS/Tabular DB, indexing some columns means to create extra data that are sorted on those columns to make a query with those columns' conditions to be scanned faster, rather than scanning from the original table data sorted based on the key only.

In NebulaGraph, the index is to create a duplicate of sorted **Vertex/Edge PROP DATA** to **locate the starting point of a QUERY**.

Not all of queries relied on the index, here are some example queries, where the starting points are only defined using conditions, rather than VertextIDs. Let's call them `pure property condition starting queries`:

```cypher
#### Queries relying on NebulaGraph Index

# query 0 pure-property-condition-start query
LOOKUP ON tag1 WHERE col1 > 1 AND col2 == "foo" \
    YIELD tag1.col1 as col1, tag1.col3 as col3;

# query 1 pure-property-condition-start query
MATCH (v:player { name: 'Tim Duncan' })-->(v2:player) \
        RETURN v2.player.name AS Name;
```

In both `query 0` and `query 1`, the pattern is to "Find VID/EDGE only based on given property conditions". 

On the contrary, the starting point is VertexID based instead in `query 2` and `query 3`:

```cypher
#### Queries not based on NebulaGraph Index

# query 2, walk query starting from given vertex VID: "player100"

GO FROM "player100" OVER follow REVERSELY \
        YIELD src(edge) AS id | \
    GO FROM $-.id OVER serve \
        WHERE properties($^).age > 20 \
        YIELD properties($^).name AS FriendOf, properties($$).name AS Team;

# query 3, walk query starting from given vertex VID: "player101" or "player102"

MATCH (v:player { name: 'Tim Duncan' })--(v2) \
        WHERE id(v2) IN ["player101", "player102"] \
        RETURN v2.player.name AS Name;
```

If we look into `query 1` and `query 3`, which shared the same condition on vertices on `tag:player`, which are both `{ name: 'Tim Duncan' }`, they are differentiated in starting points:

For `query 3`, the index is not required as the query will start from the known vertex ID in `["player101", "player102"]` and thus:
- It'll directly fetch vertex data from `v2`'s vertex IDs
- then to GetNeighbors(): walk through edges of `v2`, GetVertices() for the next hop: `v` and filter based on the property: `name`

For `query 1`, the query has to start from `v` due to no known vertex IDs were provided:
- It'll do IndexScan() first to find all vertices only with the property condition of `{ name: 'Tim Duncan' }`
- Then, GetNeighbors(): walk through edges of `v`, GetVertices() for the next hop: `v2` 

Now, we know the whole point that matters here is on **whether we know the vertexID**. And the above differences could be shown in their execution plans with PROFILE or EXPLAIN like the following:

| `query 1`, requires index(on tag: player), pure property condition query as starting point | `query 3`, no index required, query starting from known vertex IDs |
| ------------------------------------------------------------ | ------------------------------------------------------------ |
| ![query-based-on-index](https://user-images.githubusercontent.com/57335825/167237932-effed472-6390-460c-89b5-c759644dc674.png)         | ![query-requires-no-index](https://user-images.githubusercontent.com/57335825/167237957-14981b0b-61c2-4ea9-ab60-02ee49c4843b.png)   |

### Why NebulaGraph index is an enabler rather than an accelerator

Can't those queries be done without indexes?

It's possible in theory with a full scan, but disabled without an index.

The reason is that NebulaGraph stores data in a distributed and graph-oriented way, the full scan of data was considered too expensive to be allowed.

> Note: from v3.0, it's possible to do TopN Scan without index, where the `LIMIT <n>` is used, this is different from the full scan case(index is a must), which will be explained later.
>
> ```cypher
> MATCH (v:player { name: 'Tim Duncan' })-->(v2:player) \
>         RETURN v2.player.name AS Name LIMIT 3;
> ```

### Why starting point only

Index data is not used in the traversal. It could confuse us to think of index as sorting data based on properties, does it accelerate the traversal with property condition filtering? The answer is no.

In NebulaGraph, the data is structured in a way to enable fast graph-traversal, which is already indexed/sorted on vertex ID(for both vertex and edge) in the raw data, where traversal(underlying in storage, it's calling GetNeighbors interface) of the given vertex is cheap and fast due to the persistent storage.

So in summary:

> NebulaGraph Index is sorted property data to find the starting vertex or edge on given pure property conditions.

## Facts on NebulaGraph Index

To understand more details/limitations/cost of Nebula, let's reveal more about its design. Here are some facts in short:

- Index Data is stored and shared together with Vertex Data

- It's **Left Match** based only: It's RocksDB Prefix Scan under the hood

- Effect on write and read path(to see its cost):

  - Write Path: Extra Data written + Extra Read request introduced
  - Read Path: RBO(Rule-based optimization), Fan Out(to all shards)

- Data Full Scan LIMIT Sample(not full scan) is supported without index
  - `LOOKUP ON t YIELD t.name | LIMIT 1`

  - ```cypher
    MATCH (v:player { name: 'Tim Duncan' })-->(v2:player) \
            RETURN v2.player.name AS Name LIMIT 3;
    ```

The key info can be seen in one of my [sketch notes](https://www.siwei.io/en/sketch-notes/):

![nebula-graph-index-sketch](https://user-images.githubusercontent.com/57335825/167237972-507c1344-636e-48f5-9b3f-fc8ba5dbaa69.png)

> We should notice that only the left match is supported in pure-property-condition-start queries. For queries like wildcard or regular expression, Full-text Index/Search is to be used, where an external elastic search is integrated with Nebula: please check [NebulaGraph Full text index](https://docs.nebula-graph.io/3.1.0/4.deployment-and-installation/6.deploy-text-based-index/2.deploy-es/) for more.

Within this sketch note, more highlights are:

- It's a Local Index Design
  - The index is stored and shared locally together with the graph data.
  - It's sorting based on property value, and the index search is underlying a RocksDB prefix scan, that's why only left match is supported.

- There are costs in the writing path
  - The index enables the RDBMS-like property Condition Based Query with cost in the write path including not only the extra write but also, random read, to ensure the data consistency.
  - Index Data writing is done in a synchronous way

- For the reading path:
  - In pure-property-condition-start queries, in GraphD, the index will be selected with Rule-based-optimization like this example, where, in a rule, the col2 to be sorted first is considered optimal with the condition: col2 equals 'foo'.
  - After the index was chosen, the index-scan request will be fanout to storageD instances, and in the case of filters like LIMIT N, it will be pushed down to the storage side to reduce data payload.
    - Note: it was not shown in the sketch but actually from v3.0, the nebula graph allows LIMIT N Sample Property condition query like this w/o index, which is underlying pushing down the LIMIT filter to storage side.


Takeaways:

- Use index only when we have to, as it's costly in write cases and if the limit N sample is the only needed case and it's fast enough, we can use that instead.
- Index is left to match
  - composite index order matters, and should be created carefully.
  - for full-text search use case, use [full-text index](https://docs.nebula-graph.io/3.1.0/4.deployment-and-installation/6.deploy-text-based-index/2.deploy-es/) instead.

## How to use the index

We should always refer to the [documentation](https://docs.nebula-graph.io/3.1.0/3.ngql-guide/14.native-index-statements/), and I just put some highlights on this here:

- To create an index on a tag or edge type to specify a list of props in the order that we need.

  - `CREATE INDEX`

- If an index was created after existing data was inserted, we need to trigger an index asynchronously to rebuild the job, as the index data will be written in a synchronous way only when the index is created.

  - `REBUILD INDEX`

- We can see the index status after `REBUILD INDEX` is issued.

  - `SHOW INDEX STATUS`

- Queries levering index could be LOOKUP, and with the pipeline, in most cases, we will do follow-up graph-walk queries like:

  ```sql
  LOOKUP ON player \
    WHERE player.name == "Kobe Bryant"\
    YIELD id(vertex) AS VertexID, properties(vertex).name AS name |\
    GO FROM $-.VertexID OVER serve \
    YIELD $-.name, properties(edge).start_year, properties(edge).end_year, properties($$).name;
  ```

- Or in MATCH query like this, under the hood, v will be searched on index and v2 will be walked by default graph data structure without involving index.

  ```cypher
  MATCH (v:player{name:"Tim Duncan"})-->(v2:player) \
    RETURN v2.player.name AS Name;
  ```

## Recap

Finally, Let's Recap

- Index is to sort Property DATA to enable finding starting point on a given Pure Property Condition(no VertexID provided)
- Index is **not** for traversal
- Index is left match, **not** for full-text search
- Index has cost on WRITE
- Remember to REBUILD after `CREATE INDEX` on existing data

Happy Graphing!

## About the author
Wey Gu is the Developer Advocate of NebulaGraph. He is passionate about spreading the graph technology to the developer community and trying his best to make distributed graph database more accessible. Follow him on [Twitter](https://twitter.com/wey_gu) or visit his [blog](https://www.siwei.io/en/) for more fun stuff.


This article explains what NebulaGraph's native index is, and when and how to use index.

Everything you need to know about NebulaGraph Index

In this article, I will try to walk you through four different ways that you can make NebulaGraph and Apache Spark work together: Spark Connector, Nebula Exchange, Nebula Algorithm, and PySpark.

4 different ways to work with NebulaGraph in Apache Spark


![Introducing TOSS: Achieving Eventual Consistency of Edges in NebulaGraph](https://user-images.githubusercontent.com/90186547/141244981-99cea678-95fe-4929-b764-ce61e3b8520c.jpg)

NebulaGraph has just released v.2.6. In this version, TOSS is certainly one of the important features. Here a detailed explanation about TOSS will be given.

# Let's start from a GO statement

As we all know, there are two types of edges, directed and undirected edges. When traversing directed edges, you can traverse forward or reversely. NebulaGraph also supports this semantics. For example:

```golang
go from "101" over known reversely yield known.kdate, id($$);
```

The above statement starts from Vertex 101 to find all the corresponding neighbors reversely. However, when you insert an edge into NebulaGraph, the command will be like:

```golang
insert edge known(degree) VALUES "100" -> "101":(299792458);
```

Seemingly, the above statement only specifies the outgoing edge. This is because NebulaGraph will specify the incoming edge in the background when you insert an edge.

# How to insert an edge into NebulaGraph

Take the `INSERT` statement above as an example, the background execution process contains the following:

* Nebula Console sends the INSERT request to the NebulaGraph server.

* After the NebulaGraph server receives the request, it adds an incoming edge for each outgoing edge and sends `AddEdgeRequest` to their hosts respectively.

* After the host (Nebula Storage server) receives `AddEdgeRequest`, it inserts the edge locally (via the Raft protocol) and returns the result to the NebulaGraph server.

* The NebulaGraph server then returns the results from both hosts to the Nebula Console for querying purpose.

The flow diagram is as follows:

![How to insert an edge into NebulaGraph](https://user-images.githubusercontent.com/90186547/141245971-9d62e48b-7646-4d9c-a929-d8ac7d45c737.png)

If you are familiar with network/distributed programming, you may see the problem now. The graph service uses RPC to call both storage services. When the INSERT operation is executed enough times, one RPC succeeds while the other fails due to timeout. In other words, an `INSERT` operation may succeed on the outgoing edge while fail on the incoming-edge.

If now a user requires consistent property settings for both outgoing edge and incoming edge, the client has to retry the query infinitely. It is inappropriate for NebulaGraph, s a database product, to rely on the client for data atomicity.

A requirement thus comes into being, that is, to ensure the atomicity of outgoing edge and incoming edge. This means that the outgoing edge and incoming edge should be updated either successfully or they should fail at the same time. And TOSS (Transaction On Storage Side) is the feature to ensure the eventual consistency of edges upon `INSERT`, `UPDATE`, and `UPSERT` operations.

# How to use TOSS

With the release of NebulaGraph v2.6.0, the TOSS function has also been launched. The feature is set to `Disabled` by default due to performance and stability considerations. You can find the `enable_experimental_feature` option in the NebulaGraph server configuration file and set it to `True`. Then you need to restart the graphd service for the feature to take effect. The command is as follows:

```
--enable_experimental_feature=true
```

Then the operations `CREATE SPACE` / `CREATE EDGE` / `INSERT` / `UPDATE` will achieve eventual consistency of edges in NebulaGraph. (Just execute the operations as before)

Note: The TOSS feature will be only applied to incremental data after it is enabled. 


More information about v2.6, please refer to [NebulaGraph Database Manual](https://docs.nebula-graph.io/3.0.1/pdf/NebulaGraph-EN.pdf). It records in detail the knowledge points and specific usage of the graph database and the graph database NebulaGraph.

Welcome to join our [Slack channel](https://join.slack.com/t/nebulagraph/shared_invite/zt-7ybejuqa-NCZBroh~PCh66d9kOQj45g) if you want to discuss with the rest of the NebulaGraph community!

NebulaGraph has just released v2.6. In this version, TOSS is certainly one of the important features. Here a detailed explanation about TOSS will be given.

Introducing TOSS: Achieving Eventual Consistency of Edges in NebulaGraph


![Understanding Subgraph in NebulaGraph 2.0](https://user-images.githubusercontent.com/57335825/124857891-aabac500-dfdf-11eb-8461-f77ffe6f6136.png)

# Introduction

In An Introduction to [NebulaGraph 2.0 Query Engine](https://nebula-graph.io/posts/nebula-query-engine-introduction/), I introduced how the query engine differs between V2.0 and V1.0 of NebulaGraph.

![NebulaGraph Architecture Change](https://user-images.githubusercontent.com/57335825/124858026-e3f33500-dfdf-11eb-8f0f-0821af5d84b7.png)

As shown in the preceding figure, you can see that when a query statement is sent from the client, how the query engine parses the statement, generates an AST, validates the AST, and then generates an execution plan. In this article, I will introduce more about the query engine through the new subgraph feature in NebulaGraph 2.0 and focus on the execution plan generation process, to help you understand the source code better.

# Definition of Subgraph

A subgraph of a graph is composed of a subset of the graph's vertices and a subset of its edges. When a starting vertex is given, a traversal starts from it and expands following the specified edge type until all the vertices at the specified number of hops are reached. All the reachable vertices and edges form a subgraph. 

# Syntax of Subgraph

```shell
GET SUBGRAPH [<step_count> STEPS] FROM {<vid>, <vid>...} [IN <edge_type>, <edge_type>...]
[OUT <edge_type>, <edge_type>...] [BOTH <edge_type>, <edge_type>...]
```

- step_count: Specifies the number of hops from the starting vertex to obtain a subgraph consisting of vertices and edges that are reached at 0 to `step_count` hops. It must be a non-negative integer. The default value is 1.
- vid: Specifies the ID of the starting vertex.
- edge_type: Specifies the edge type. `IN`, `OUT`, and `BOTH` can be added to specify the direction of the edges from the starting vertex. By default, `BOTH` is used.

# Implementation of Subgraph

When the query engine receives a `GET SUBGRAPH` statement, the Parser, composed of Flex and Bison, will extract the necessary content from the query statement according to the specified rule, that is, the `get_subgraph_sentence` rule defined in [parser.yy](https://github.com/vesoft-inc/nebula-graph/blob/master/src/parser/parser.yy), and then generate an AST, as shown in the following figure.

![NebulaGraph Subgraph Implementation](https://user-images.githubusercontent.com/57335825/124858306-5e23b980-dfe0-11eb-8f66-bd15790a0eef.png)

And then in the Validate stage, the generated AST is validated, aiming to make sure that all the inputs from users conform to the syntax rule. For more information, see [An Introduction to NebulaGraph 2.0 Query Engine](https://nebula-graph.io/posts/nebula-query-engine-introduction/). When the validation is successful, the content of the AST will be extracted and an execution plan will be generated.

How is the execution plan generated? Database solutions may differ in the form of the execution plan, but their implementations are the same, that is, depending on which operators are used and how the query layer interacts with the storage layer, because data needs to be taken from the storage layer for each query statement. In NebulaGraph, the query engine interacts with the storage layer through RPC (fbthrift), of which the interfaces are defined in the `interface` directory of the `common` repository. `getNeighbors` and `getProps` are the key interfaces.

In fbthrift, `getNeighbors` is defined as follows:

```shell
struct GetNeighborsRequest {    
    1: common.GraphSpaceID                      space_id,
    2: list<binary>                             column_names,
    3: map<common.PartitionID, list<common.Row>>
        (cpp.template = "std::unordered_map")   parts,
    4: TraverseSpec                             traverse_spec
}
```

To understand the variables in the preceding structure, see the comments in the source code: [https://github.com/vesoft-inc/nebula-common/blob/master/src/common/interface/storage.thrift](https://github.com/vesoft-inc/nebula-common/blob/master/src/common/interface/storage.thrift).

It mainly aims to enable this process:

1. The query engine passes the starting vertex and the specified edge types according to the defined structure.
2. In the storage layer, the starting vertex is located.
3. The properties of the vertex and its outgoing edges are assembled into a table and the table is returned to the query engine. The table format is defined in `GetNeighborsResponse`. For more information, see [https://github.com/vesoft-inc/nebula-common/blob/master/src/common/interface/storage.thrift](https://github.com/vesoft-inc/nebula-common/blob/master/src/common/interface/storage.thrift).
4. The query engine extracts the necessary content from the table.

For example, for the NBA dataset, when `Tim Duncan` and `Manu Ginobili` are given as the starting vertices and `like` of both directions are given as the edge type, we can get the information of `$^.player.name`, `like._dst`, `$$.player.name`, and `like.likeness` as follows:

![NebulaGraph Subgraph](https://user-images.githubusercontent.com/57335825/124858735-32ed9a00-dfe1-11eb-8bb1-057bed6ac316.png)

Because the traversal is done in both directions, `+ like` in the fourth column of the preceding table represents outgoing edges, and `- like` in the fifth column represents incoming edges.

In NebulaGraph, edges are stored with their source vertices in the storage layer, so all the properties of a vertex and its outgoing edges can be obtained through the `GetNeighbor` interface. However, if you want to obtain the property information of the destination vertices of these edges, the `getProps` interface is needed. A `FETCH` statement also needs this interface to obtain the property information of a vertex or an edge. For more information about `GetPropRequest`, see [https://github.com/vesoft-inc/nebula-common/blob/master/src/common/interface/storage.thrift](https://github.com/vesoft-inc/nebula-common/blob/master/src/common/interface/storage.thrift).

## Execution Plan

With the preceding interfaces, an execution plan can be generated. The operators, `start`, `getNeighbor`, `subgraph`, `loop`, and `dataCollect`, are necessary.

- start: A leaf node of an execution plan. It does nothing but informs the scheduler that no more operator can be relied upon. In other words, this is the termination condition in the recursive algorithm.
- loop: Equivalent to the `WHILE` loop in C language. It is composed of three parts: `depend`, `condition`, and `loopBody`. `depend` is used in multi-statements and the PIPE and no more details about it is involved in this article. `condition` is equivalent to the termination condition. `loopBody` is equivalent to the loop body in a `WHILE` loop.
- subgraph: Responsible for extracting the properties of `_dst`, the destination vertex, from the `getNeighbor` result, filtering out the destination vertices that have been traversed to avoid repeatedly fetching data from the storage layer, and then using them as the input for the next hop executed by `getNeighbor`.
- datacollect: Responsible for assembling the properties of the reachable vertices and their edges into the vertices and edge types.

For more information about these operators, see the source code: [https://github.com/vesoft-inc/nebula-graph/tree/master/src/executor](https://github.com/vesoft-inc/nebula-graph/tree/master/src/executor).

Let's take the graph in the following figure as an example to introduce how a subgraph is generated.

![Build a Subgraph in NebulaGraph](https://user-images.githubusercontent.com/57335825/124859204-0a19d480-dfe2-11eb-8a1d-c83af10038fe.png)

**Single-hop traversal**

It is easy to start from Vertex A and traverse along the `like` edges at one hop for the vertices and edges. In this scenario, only getNeighbor and dataCollect are needed. The execution plan is as follows.

![Single-hop traversal](https://user-images.githubusercontent.com/57335825/124859285-37668280-dfe2-11eb-9021-ab538e2ac90b.png)

**Multi-hop traversal**

The single-hop traversal is a special case of the multi-hop traversal, so it can be treated as part of a multi-hop traversal. In a three-hop traversal starting from Vertex A along the `like` edges at three hops, the destination vertices can be extracted after `getNeighbor` is executed and these vertices can be the starting vertices to call the `getNeighbor` interface. Two loops are sufficient for such a scenario. The number of hops can be set as the termination condition in `loop`. Therefore, the execution plan is as follows.

![Multi-hop traversal](https://user-images.githubusercontent.com/57335825/124859410-77c60080-dfe2-11eb-836e-3450dfa005bd.png)

**Input and output**

Generally, the input of each operator is the output of the one that it depends on, so according to the dependencies represented in the execution plan, you can see the input and output of each operator. However, in some cases, for example, in a multi-hop traversal, the input of each `getNeighbor` is the destination vertex that is reached following an edge, that is, the output of the `subgraph` operator, so the output of `subgraph` is the input of `getNeighbor`. Such a dependency is different from that shown in the preceding figure. Therefore, the input and output of each operator must be set separately. From [An Introduction to NebulaGraph 2.0 Query Engine](https://nebula-graph.io/posts/nebula-query-engine-introduction/), you know that the input and output of each operator are stored in a hash table, of which values are of the `vector` type. The following ResultMap table is an example.

![ResultMap](https://user-images.githubusercontent.com/57335825/124859705-f4f17580-dfe2-11eb-8e9e-5b7bbadc5686.png)

- The starting vertex is stored in ResultMap["StartVid"].
- The input of getNeighbor is ResultMap["StartVid"] and its output is stored in ResultMap["GN_1"].
- The input of getNeighbor is ResultMap["StartVid"] and its output is stored in ResultMap["GN_1"].
- No data is generated by loop. It is used as a logic loop, so neither input nor output is necessary.
- The input of dataCollect is ResultMap["GN_1"] and its output is stored in ResultMap["DATACOLLECT_2"].

At this time, the getNeighbor operator puts the result of each loop at the end of the vector in ResultMap["GN_1"], and the subgraph operator takes the value from the end of the vector in ResultMap["GN_1"] and then stores the starting vertex for the next hop in ResultMap["StartVid"].

After the first hop, the result of ResultMap is as follows.
​
![ResultMap](https://user-images.githubusercontent.com/57335825/124859885-3eda5b80-dfe3-11eb-8eff-e0705c0404e9.png)

To make the result clearer, only the properties of `_dst` are shown as the result of `getNeighbor`. In the actual result, the properties of all the edges and the starting vertex are listed, as show in Table 1.

`subgraph` receives "GN_1" as its input, extracts the properties of `_dst`, and then stores the result in "StartVid".

After the second hop traversal, the ResultMap is as follows.

![ResultMap](https://user-images.githubusercontent.com/57335825/124860033-882aab00-dfe3-11eb-8237-0987c712006a.png)

After the third hop traversal, the ResultMap is as follows.

![ResultMap](https://user-images.githubusercontent.com/57335825/124860085-a2fd1f80-dfe3-11eb-8f14-54aaaa0950ca.png)

Finally, `dataCollect` takes all the vertex sets and edge sets traversed in the process, assembles them into the final result, and returns it for users.

# Example

In this section, I will use an example to show the execution plan structure of a subgraph in NebulaGraph 2.0:

1. Open Nebula Console.
2. Switch to the `nba` graph space. 
3. Run `EXPLAIN format="dot" GET SUBGRAPH 2 STEPS FROM 'Tim Duncan' IN like, serve`. Data of the DOT format is generated on Nebula Console.
4. Open the [Graphviz Online](https://dreampuf.github.io/GraphvizOnline/) website and copy and paste the DOT data on it.

The execution plan is shown as follows.

![Result in DOT format](https://user-images.githubusercontent.com/57335825/124860296-fc654e80-dfe3-11eb-86fc-589ed9a3babe.png)

In this plan, `depend` of `loop` depends on `Start_0`. However, because there is no multi-statement or PIPE, it is not necessary.
​
Welcome to join the NebulaGraph [Slack](https://join.slack.com/t/nebulagraph/shared_invite/zt-7ybejuqa-NCZBroh~PCh66d9kOQj45g) channel to learn more about NebulaGraph.


This article introduces the query engine, the execution plan generation process in particular, through the new subgraph feature in NebulaGraph 2.0 to help you understand the source code better.

Understanding Subgraph in NebulaGraph 2.0

NebulaGraph 2.0 supports full-text indexing by using an external full-text search engine. To understand this new feature, let's review the architecture and storage model of NebulaGraph 2.0.

Full-Text Indexing in NebulaGraph 2.0


![storage-format-in-nebula-graph-2.0](https://user-images.githubusercontent.com/57335825/118632761-5119f380-b803-11eb-8ab4-ca4e8a7e7bf6.png)

NebulaGraph 2.0 has changed a lot over its releases. In the storage architecture design, the encoding format has been changed, which has the most significant impact on its users. In NebulaGraph, data is stored as KV-pairs in RocksDB. This article covers several issues such as the differences between the old and new encoding formats and why the format must be changed.

# Encoding Format in NebulaGraph 1.0

Let's start with a brief review of the encoding format in NebulaGraph 1.0. For those who are not familiar with it, I recommend that they read this post: [An Introduction to NebulaGraph's Storage Engine](https://nebula-graph.io/posts/nebula-graph-storage-engine-overview/). In NebulaGraph 1.0, the vertex IDs can only be represented by values of the `int` type, so all VertexIDs are stored in int64.

* Vertex key format in NebulaGraph 1.0

![Vertex key format in NebulaGraph 1.0](https://user-images.githubusercontent.com/57335825/118632935-8292bf00-b803-11eb-8f97-08f1207971fe.png)

* Edge key format in NebulaGraph 1.0

![Edge key format in NebulaGraph 1.0](https://user-images.githubusercontent.com/57335825/118633021-9a6a4300-b803-11eb-9d33-3f9743938b84.png)

For a VertexID, its corresponding PartID is obtained after hash, so a vertex and all its incoming and outgoing edges, of which only the IDs of the sourcing vertices are  hash processed, are mapped in the same partition. **It should be noted that in NebulaGraph 1.0, the first bytes of a vertex and an edge are of the same Type**. That is, for a vertex, all its tags are not stored physically contiguously. They may be stored as shown in the following figure. For the vertex named `src`, its three tags (tag1, tag2, and  tag3) may actually be separated by edges.

![rocksdb key format](https://user-images.githubusercontent.com/57335825/118633166-c38ad380-b803-11eb-8fb6-bc8d7f83a95c.png)

Such a key format can meet most requirements for operations on data in NebulaGraph 1.0. For example, when the prefix is specified, data can be retrieved by the `FETCH` and the `GO` manipulations.

# Encoding Format in NebulaGraph 2.0

In the releases prior to the 2.0 GA release, the encoding format in the underlying storage layer was actually the same as that in NebulaGraph 1.0. For an integer VertexID, it is stored in exactly the same format as in 1.0. However, for a string VertexID, it is changed from int64, which occupies 8 bytes, to `FIXED_STRING` of a fixed length which needs to be specified by users when they run `CREATE SPACE`. If a VertexID is shorter than the specified length, the system will automatically pad it with `\0`. For a VertexID that exceeds the specified length, the system will report an error directly.

In the 2.0 GA release, several changes were made to the underlying storage encoding format. Therefore, to upgrade NebulaGraph to this release, a specific tool is necessary. This tool converts the data from the old format to the new format. The following figures show the key formats implemented in the 2.0 GA release.

## Key Formats in NebulaGraph 2.0

* Vertex key format

![Vertex key format](https://user-images.githubusercontent.com/57335825/118633434-0fd61380-b804-11eb-98fb-93e80c0fbc5c.png)

* Edge key format

![Edge key format](https://user-images.githubusercontent.com/57335825/118633579-35631d00-b804-11eb-835e-32928e22e179.png)

## Key Format Comparison between NebulaGraph 1.0 and 2.0

![Key Format Comparison - Vertex](https://user-images.githubusercontent.com/57335825/118633663-47dd5680-b804-11eb-944a-83ecb909c154.png)

![Key Format Comparison - Edge](https://user-images.githubusercontent.com/57335825/118633757-63486180-b804-11eb-9b9f-2534d76be91f.png)

As shown in the preceding figures, some major changes are made:

1. As mentioned earlier, the length of VertexID is changed from 8 bytes to n bytes. For an integer VertexID, n takes the value of 8. For a string VertexID, n takes the value of the specified length.
2. In the vertex key format, the timestamp byte is removed, and in the edge key format, the timestamp byte is replaced with a one-byte placeholder.
3. For the key format of a vertex and an edge, **the Type byte differs from each other**, which separates vertices and edges physically. 

Why these changes are made? Here are the main reasons: 

1. The changes made to VertexID are mainly to make it support string IDs while still being compatible with int IDs in NebulaGraph 1.0. In the Storage layer, VertexIDs are encoded as bytes. However, VertexIDs of the corresponding type, which is determined by the configuration of the graph space, are returned in the results. 
    > Why FIXED_STRING is applied to the string IDs? If a fixed length is not specified, scanning by prefixes is not possible. By padding the length with \0, the prefixes of all vertices and edges are of the same length, so that the corresponding prefix query can be performed.
2. Removing the timestamp byte is mainly to improve the performance, because saving data as multiple versions lowers the performance and currently the MVCC implementation in NebulaGraph is of a low priority.
    > A one-byte long placeholder is reserved in the edge key format, which is purposely designed for TOSS (Transaction on Storage Side). It is mainly used to identify the outgoing edge and the incoming edge of an edge. We will write another article to introduce the TOSS feature in detail. Please stay tuned for our updates.
3. The main benefit of separating vertices and edges physically is that all tags of a vertex can be easily and quickly retrieved, which is extensively used in Cypher's `MATCH` statements. Before this change, data is scanned by a prefix composed of the same `Type + VertexID`, which may greatly reduce the performance because vertices and edges may be mixed together. After different Types are implemented, scanning by the VertexType + VertexID prefix can retrieve all tags quickly.
    > In NebulaGraph 1.0, vertices and edges can be stored with the same prefix because there is no need to fetch all tags for a vertex. At the code level, however, such an implementation has great effect on the performance. For example, the FETCH operation in NebulaGraph 1.0 works by scanning by the VertexID prefix, but for a super vertex, retrieving all its tags with such an operation performs poorly. Besides, if the SCAN operation provided by the Storage layer is used to retrieve all the vertices of an entire graph, the entire RocksDB is actually scanned.

In addition to the changes made to the key format of vertices and edges, the format of the indexes has actually changed. 

Why these changes? The first reason is that with `NULL` support in NebulaGraph 2.0, the index semantics must be changed accordingly. The second one is that in NebulaGraph 1.0, the string fields in indexes are actually treated as strings of variable length. Therefore, whenever an index on a string field is used in a `LOOKUP` statement, only an equivalent query is used. In NebulaGraph 2.0, however, `FIXED_STRING` is applied to the string fields for indexes and the data VertexID, and for the `LOOKUP` statements, range scanning is supported for the string type index. For example, `LOOKUP ON index1 WHERE col > "aaa"`. We will introduce this design in details in the future articles.

Interested in this topic? Join the NebulaGraph [Slack channel](https://join.slack.com/t/nebulagraph/shared_invite/zt-7ybejuqa-NCZBroh~PCh66d9kOQj45g) and talk to the community.


This article covers several issues such as the differences between the old and new encoding formats and why the format must be changed.

Storage Format in NebulaGraph v2.0.0


In the process of using NebulaGraph, users often encounter various problems. To troubleshoot them, we usually recommend running the SHOW HOSTS statement first to view the cluster status. In a sense, the status of a NebulaGraph cluster is determined by a heartbeat mechanism. 

This article introduces the heartbeat mechanism through which the servers in a NebulaGraph cluster communicate with each other.

# What is Heartbeat?

![Heartbeat Mechanism in NebulaGraph](https://user-images.githubusercontent.com/57335825/114685585-99e21680-9d44-11eb-8cc9-6799fe490560.png)

A NebulaGraph cluster contains three types of servers: nebula-graphd servers for querying, nebula-storaged servers for storing business data, and nebula-metad servers for storing metadata. The heartbeats mentioned in this article refer to the heartbeat messages periodically sent by nebula-graphd servers and nebula-storaged servers to nebula-metad servers. With the heartbeats, a cluster can perform the following functions. (The necessary parameter is `heartbeat_ interval_ secs`.)

> The heartbeats used by Raft, often mentioned in NebulaGraph, refer to the heartbeats between the nebula-storaged servers with the same partition, which is different from the heartbeats referred to in this article.

## 1. Discovering Services

Before starting a NebulaGraph cluster, users must configure the `meta_host_addrs` parameter in the specified configuration file. When the nebula-graphd servers and the nebula-storaged servers start, they send heartbeat messages to the nebula-metad servers at the specified `meta_host_addrs` addresses. External access to the nebula-graphd servers and the nebula-storaged servers is not available until they connect to the nebula-metad servers. When the nebula-metad servers receive the heartbeat messages, they store the necessary metadata (see the "Reporting Information of servers" section). From then, users can view the connected nebula-storaged servers by running the `SHOW HOSTS` statement. For NebulaGraph 2.x, users can list the connected nebula-graphd servers by running the `SHOW HOSTS GRAPH`statement.

## 2. Reporting Information of Cluster Nodes

When heartbeat messages are received, nebula-metad servers collect the IP address, the port number, the server type, the heartbeat time, and other information from the messages and store them for subsequent use. For more information, see the "What Do the Heartbeats do?" section.

In addition, the leader information of nebula-storaged servers is also reported when the leader number is updated. In the result returned for a `SHOW HOSTS` query, the Leader count and Leader distribution information are from the heartbeat messages.

## 3. Updating Metadata

The heartbeat response sent by nebula-metad servers to nebula-graphd servers and nebula-storaged servers contain the `last_update_time` information, which informs the last update time of the metadata. When the information is updated, nebula-graphd servers and nebula-storaged servers will send a heartbeat request to nebula-metad servers for the updated information of the graph space, the schema, and the configurations. 

> For example, when a new tag is created, before the information of the new tag is received by the nebula-graphd servers and the nebula-storaged servers, users cannot insert vertices, but receive an error like No schema found. However, after the corresponding information is obtained through heartbeat messages and stored in the local cache, users can insert vertices successfully.

# What Do the Heartbeats Do?

- Results returned for statements such as `SHOW HOSTS` and `SHOW PARTS` are combinations of the information of the heartbeat messages from all the servers that are stored in nebula-metad servers.
- Operation and maintenance statements such as `BALANCE DATA` and `BALANCE LEADER` need to retrieve the online nebula-storaged servers in a cluster. The server status is determined by nebula-metad servers according to whether the last heartbeat time is within the threshold range.
- When users create a graph space by running `CREATE SPACE`, nebula-metad servers also need the status of nebula-storaged servers and then distribute the partitions of the graph space across the online nebula-storaged servers.

> Let's take a common problem as an example: we start a nebula-storaged process, stop it, change the port number, and then start the process again. If the procedure is fast enough, we can see two online nebula-storaged servers by running `SHOW HOSTS`. If we immediately create a new graph space by running `CREATE SPACE test(partition_num=10, replica_factor=1)`, the `test` graph space will be distributed on the two nebula-storaged servers. However, if we create the graph space when only one nebula-storaged server is online and then start the offline server again, the graph space exists in only one nebula-storaged server, that is, the online one during the process of creating the `test` graph space.

![Online](https://user-images.githubusercontent.com/57335825/114686576-84212100-9d45-11eb-9a1e-e359c7c67bda.png)

![Offline](https://user-images.githubusercontent.com/57335825/114686663-9733f100-9d45-11eb-84a1-33bfc26d2f93.png)

# Evolution of Heartbeat Mechanism

In 2018-2019, the heartbeat mechanism was not so satisfactory. First, the metadata is obtained from nebula-metad servers regardless of whether the data has been updated. However, the metadata is not updated frequently, so it is a waste of resources to obtain metadata periodically. Second, to add or delete a nebula-storaged server by running statements like `ADD HOSTS` and `DELETE HOSTS`, a trustlist-like mechanism is applied. For the servers that are not authenticated, external access to them is not available. This mechanism has its own advantages, but it is not user-friendly enough, which is its biggest disadvantage.

Therefore, since the end of 2019, with the special contributions of @zhangguoqing at JD Digit, we have made some changes to the heartbeat mechanism. After a period of verification, the enhanced heartbeat mechanism is just what you see now.

# Extra

Regarding the heartbeat mechanism, you should pay attention to the _cluster.id_ file. It is designed to prevent nebula-storaged servers from communicating with the wrong nebula-metad servers. Here is how it is implemented:

- First, when nebula-metad servers start, their hash values are generated based on the configuration of `meta_host_addrs` and are stored as KV pairs locally.
- When nebula-storaged servers start, they will try to obtain the hash values of the corresponding nebula-metad servers from the _cluster.id_ file, and send them in the heartbeat message. If the file does not exist, 0 is used for the first time instead. When heartbeat response are received, the hash values are stored in the _cluster.id_ file and the values will be used all the time.
- During the processing of heartbeats, nebula-metad servers compare the local hash values with those in a heartbeat request sent by nebula-storaged servers. If they do not match, the request is rejected. In this case, external access to nebula-storaged servers is not available. This is how the _Reject wrong cluster host_ log is generated.

This is an introduction to the heartbeat mechanism of NebulaGraph. If you are interested, see how it is implemented in the source code: [https://github.com/vesoft-inc/nebula-graph](https://github.com/vesoft-inc/nebula-graph).

# You Might Also Like

- [How Indexing Works in NebulaGraph](https://nebula-graph.io/posts/how-indexing-works-in-nebula-graph/)
- [An Introduction to Snapshot in NebulaGraph](https://nebula-graph.io/posts/nebula-graph-snapshot-introduction/)



This article introduces the heartbeat mechanism through which the servers in a NebulaGraph cluster communicate with each other.

Communications in a Cluster: About Heartbeat Mechanism


![nebula-spark-connector-writer](https://user-images.githubusercontent.com/57335825/113999158-5b53e400-988c-11eb-9606-6f9217fedf48.png)

In "[Nebula Spark Connector Reader: Implementation and Practices](https://nebula-graph.io/posts/nebula-spark-connector-reader/)", it is known that Nebula Spark Connector is a connector for Spark reading data from or writing data into NebulaGraph. It is composed of Reader and Writer. In this article, I will introduce how Nebula Spark Connector makes writing data from other data sources into NebulaGraph possible. 

# How Nebula Spark Connector Writer is Implemented

Spark SQL supports customizing data sources, making Spark integrating with third-party data sources possible. 

Writing a single record into NebulaGraph with Nebula Spark Connector is implemented based on [DatasourceV2](https://spark.apache.org/docs/2.3.0/api/java/index.html?org/apache/spark/sql/sources/v2/DataSourceV2.html). Here is how it is implemented:

1. The `WriteSupport` interface is inherited and a new `createWriter` method is defined to create a custom `DataSourceWriter`.
2. The `DataSourceWriter` interface is inherited and the `NebulaDataSourceVertexWriter` class and the `NebulaDataSourceEdgeWriter` class are created. A new `createWriterFactory` method is defined and a custom `DataWriterFactory` is returned. A new `commit` method is created to commit the entire transaction. A new abort method is defined for transaction rollback. Please note that NebulaGraph 1.x does not support transactions, so the `commit` and `abort` methods in this implementation will not be actually called.
3. The `DataWriterFactory` is inherited and the `NebulaVertexWriter` class and the `NebulaEdgeWriter` class are created. The `createWriter` method is defined and a custom `DataWriter` is returned. 
4. The `DataWriter` interface is inherited and the `NebulaVertexWriter` class and the `NebulaEdgeWriter` class are created. A new `write` method is defined to write data out. A new `commit` method is created to commit the entire transaction. A new `abort` method is defined for transaction rollback. Please note that NebulaGraph 1.x does not support transactions, so the `commit` and `abort` methods in `DataWriter` will not be actually called.

The class diagram of the implementation of Nebula Spark Connector Writer is as follows.

![nebula-spark-connector-writer-diagram](https://user-images.githubusercontent.com/57335825/113999974-21371200-988d-11eb-96d0-00e33a760566.png)

How to write a single record into NebulaGraph is implemented in the `write` method in `NebulaVertexWriter` and `NebulaEdgeWriter`: 

1. Creates a client and connects it to the Graph Service of NebulaGraph.
1. Specifies a graph space to write data into.
1. Constructs nGQL statements to insert data into NebulaGraph.
1. Commits the nGQL statements and executes them.
1. Defines callback functions to receive the execution results. 

Nebula Spark Connector Writer batch writes data into NebulaGraph via cumulative commits of batch data transformed by the `map` operation on DataFrame, which is similar to the implementation of [Nebula Exchange](https://docs.nebula-graph.io/1.1/nebula-exchange/about-exchange/ex-ug-what-is-exchange/). 

# Practices of Nebula Spark Connector Writer

Nebula Spark Connector Writer provides users with two interfaces to program to write data into a third-party data source. The source of the data is DataFrame. Nebula Spark Connector Writer provides two types of interfaces to write single records and write data in batches.

Run these commands line by line to pull the source code of Nebula Spark Connector from GitHub and compile Nebula Spark Connector. Then, copy the package to the local Maven repository.

```shell
git clone -b v1.0 https://github.com/vesoft-inc/nebula-java.git
cd nebula-java
mvn clean install -Dgpg.skip -Dmaven.javadoc.skip=true -Dmaven.test.skip=true
```

Here is an example to introduce how to use Nebula Spark Connector Writer.

1. Add the `nebula-spark` dependency into the POM file of your Maven project. 

```shell
<dependency>
  <groupId>com.vesoft</groupId>
  <artifactId>nebula-spark</artifactId>
  <version>1.0.1</version>
</dependency>
```

2. The Spark application uses an interface of Nebula Spark Connector Writer to write data of DataFrame into NebulaGraph.

- 2.1 Writes a single record into NebulaGraph: 

```shell
// Constructs DataFrame for vertices and edges. 
// The demo data in the nebula-java/examples/src/main/resources directory of the v1.0 branch of the nebula-java repository is used. 
// You can find the demo data in the local examples/src/main/resources directory.

val vertexDF = spark.read.json("examples/src/main/resources/vertex")
    vertexDF.show()
val edgeDF = spark.read.json("examples/src/main/resources/edge")
        edgeDF.show()

// Writes a vertex
vertexDF.write
  .nebula("127.0.0.1:3699", "nb", "100")
  .writeVertices("player", "vertexId", "hash")
  
// Writes an edge
edgeDF.write
    .nebula("127.0.0.1:3699", "nb", "100")
  .wirteEdges("follow", "source", "target")

```
Explanation of the parameters: 

   - nebula(address: String, space: String, partitionNum: String)
      - address：Specifies the Graph server addresses and their ports. Multiple addresses are supported. The addresses must be separated with commas. The default port is 3699. For example, "ip1:3699,ip2:3699".
      - space: Specifies the graph space name of NebulaGraph.
      - partitionNum：Sets to the value of partitionNum that was specified during the creation of the graph space. If no value was specified, the default value `100` is used.
   - writeVertices(tag: String, vertexFiled: String, policy: String = "")
      - tag：Sets to a tag name created in the graph space. 
      - vertexFiled：Specifies a column of the DataFrame as the source of the vertex IDs. For example, if the DataFrame is composed of three columns, a, b, and c, and column a is used as the source of the vertex IDs, then set this parameter to a.
      - policy：If data of the `vertexField` column is not of an `int` type, a mapping policy is necessary and set policy to "hash". Otherwise, leave it blank.
   - writeEdges(edge: String, srcVertexField: String, dstVertexField: String, policy: String = "")
      - edge：Sets to an edge type name created in the graph space.
      - srcVertexField：Specifies a column of the DataFrame as the source of the source vertex ID of an edge.
      - dstVertexField：Specifies a column of the DataFrame as the source of the destination vertex ID of an edge.
      - policy：If data of the `srcVertexField` column and/or the `dstVertexField` column is not of an `int`
       type, a mapping policy is necessary and set policy to "hash". Otherwise, leave them blank.

- 2.2 Batch writes data to NebulaGraph

```shell
// Constructs DataFrame for vertices and edges. 
// The demo data in the nebula-java/examples/src/main/resources directory of the v1.0 branch of the nebula-java repository is used. 
// You can find the demo data in the local examples/src/main/resources directory.

val vertexDF = spark.read.json("examples/src/main/resources/vertex")
    vertexDF.show()
val edgeDF = spark.read.json("examples/src/main/resources/edge")
        edgeDF.show()

// Batch writes vertices
new NebulaBatchWriterUtils()
      .batchInsert("127.0.0.1:3699", "nb", 2000)
      .batchToNebulaVertex(vertexDF, "player", "vertexId")

// Batch writes edges
new NebulaBatchWriterUtils()
      .batchInsert("127.0.0.1:3699", "nb", 2000)
      .batchToNebulaEdge(edgeDF, "follow", "source", "target")

```

Explanation of the parameters: 

   - batchInsert(address: String, space: String, batch: Int = 2000)
      - address：Specifies the Graph server addresses and their ports. Multiple addresses are supported. The addresses must be separated with commas. The default port is 3699. For example, "ip1:3699,ip2:3699".
      - space：Specifies the graph space name of NebulaGraph.
      - batch：Sets to the number of records to be written into NebulaGraph in one batch. Optional. The default value is 2000 .
   - batchToNebulaVertex(data: DataFrame, tag: String, vertexField: String, policy: String = "")
      - data：Sets to the DataFrame to be written into NebulaGraph.
      - tag：Sets to a tag name created in the graph space.
      - vertexField：Specifies a column of the DataFrame as the source of the vertex IDs.
      - policy：If data of the `vertexField` column is not of an `int` type, a mapping policy is necessary and set policy to "hash". Otherwise, leave it blank.
   - batchToNebulaEdge(data: DataFrame,  edge: String, srcVertexField: String, dstVertexField: String, rankField: String = "",  policy: String = "")
      - data：Sets to the DataFrame to be written into NebulaGraph.
      - edge：Sets to an edge type name created in the graph space.
      - srcVertexField：Specifies a column of the DataFrame as the source of the source vertex ID of an edge.
      - dstVertexField：Specifies a column of the DataFrame as the source of the destination vertex ID of an edge.
      - rankField：Specifies a column of the DataFrame as the source of the rank value of an edge. Optional.
      - policy：If data of the `srcVertexField` column and/or the `dstVertexField` column is not of an `int` type, a mapping policy is necessary and set policy to "hash". Otherwise, leave them blank.

So far, the introduction to Nebula Spark Connector Writer is done. You are welcome to pull the source code from the GitHub repository ([https://github.com/vesoft-inc/nebula-java/tree/v1.0/tools/nebula-spark](https://github.com/vesoft-inc/nebula-java/tree/v1.0/tools/nebula-spark)) and try it out. 

If you encounter any problems during the trial, please feel free to ask questions and discuss in under the nebula-spark-connector  [repository](https://github.com/vesoft-inc/nebula-spark-connector/issues).

This article introduces to you how Nebula Spark Connector makes writing data from other data sources into NebulaGraph possible. 

Nebula Spark Connector Writer: Implementation and Practices


![An Introduction to NebulaGraph 2.0 Query Engine](https://user-images.githubusercontent.com/57335825/111279545-c20f2480-8675-11eb-8699-a8429c237dec.png)

# Introduction

Compared with V1.0, NebulaGraph, a distributed graph database, has been significantly changed in V2.0. One of the most obvious changes is that in NebulaGraph 1.0, the code of the Query, Storage, and Meta modules are placed in one repository, while from NebulaGraph 2.0 onwards, these modules are placed in three repositories:

* `nebula-graph`: Mainly contains the code of the Query module.
* `nebula-common`: Mainly contains expression definitions, function definitions, and some public interfaces.
* `nebula-storage`: Mainly contains the code of the Storage and Meta modules.

This article introduces the overall structure of the Query layer and uses an nGQL statement to describe how it is processed in the four main modules of the Query layer.

# Architecture

This figure shows the architecture of the Query layer. It has four submodules:

* Parser: Performs lexical analysis and syntax analysis.
* Validator: Validates the statements.
* Planner: Generates and optimizes the execution plans.
* Executor: Executes the operators.

![The architecture of the Query layer](https://user-images.githubusercontent.com/57335825/111279912-2af69c80-8676-11eb-8377-492a0955c3cd.png)

# Source Code Hierarchy

Now, let's look at the source code hierarchy under the`nebula-graph`repository.

```plain
|--src
    |--context     // contexts for validation and execution
    |--daemons 
    |--executor    // execution operators
    |--mock
    |--optimizer   // optimization rules
    |--parser      // lexical analysis and syntax analysis
    |--planner     // structure of the execution plans
    |--scheduler   // scheduler
    |--service
    |--util        // basic components
    |--validator   // validation of the statements
    |--visitor
```

#  How a Query is Executed

Starting from the release ofNebulaGraph 2.0-alpha, nGQL gets to support vertex IDs of the`String`type. It is working to support vertex IDs of the`Int`type in NebulaGraph 2.0 to get compatible with NebulaGraph 1.0.

To introduce how the Query layer works, let's take an nGQL statement like`GO FROM "Tim" OVER like WHERE like.likeness > 8.0 YIELD like._dst`as an example. Its data flow through the Query layer is shown in the following figure.

![How a Query is Executed](https://user-images.githubusercontent.com/57335825/111281267-b886bc00-8677-11eb-91ba-fcbf3371ce06.png)

When the statement is input, it is processed as follows.

## Stage 1: Generating an AST

In the first stage, the statement is parsed by the Parser (composed of Flex and Bison) and its corresponding AST is generated. The structure of the AST is shown in the following figure.

![Generating an AST](https://user-images.githubusercontent.com/57335825/111280086-5ed1c200-8676-11eb-98fa-86ba7029a7a9.png)

In this stage, the Parser intercepts the statements that do not conform to the syntax rules. For example, a statement like`GO "Tim" FROM OVER like YIELD like._dst`will be directly intercepted in this stage because of its invalid syntax.

## Stage 2: Validation

In the second stage, the Validator performs a series of validations on the AST. It mainly works on these tasks:

* Validating metadata

When parsing the `OVER` , `WHERE` , and `YIELD` clauses, the Validator looks up the Schema and verifies whether the edge and tag data exist or not. For an `INSERT` statement, the Validator verifies whether the types of the inserted data are the same as the ones defined in the Schema.

* Validating contextual reference

For multiple statements, like `$var = GO FROM "Tim" OVER like YIELD like._dst AS ID; GO FROM $var.ID OVER serve YIELD serve._dst`, the Validator verifies `$var.ID`: first to see if `var` was defined, and then to check if the `ID` property is attached to the `var` variable. If `$var.ID` is replaced with `$var1.ID` or `$var.IID` , the validation fails.

* Validating type inference

The Validator infers what type the result of an expression is and verifies the type against the specified clause. For example, the `WHERE` clause requires the result to be a Boolean value, a NULL value, or empty.

* Replacing `*`

Take a statement like `GO FROM "Tim" OVER * YIELD like._dst, like.likeness, serve._dst` as an example. When verifying the `OVER` clause, the Validator needs to query the Schema and replace `*` with all the edges defined in the Schema. For example, if only `like` and `serve` edges are defined, the statement is as follows.

```plain
GO FROM "Tim" OVER serve, like YIELD like._dst, like.likeness, serve._dst
```

* Validating input and output

For an nGQL statement with the PIPE operator, such as`GO FROM "Tim" OVER like YIELD like._dst AS ID | GO FROM $-.ID OVER serve YIELD serve._dst`, the Validator verifies `$-.ID` . In the example statement, the `ID` property in the second clause was already defined in the previous clause, so the second clause is verified legal. If `$-.ID` is changed to `$-.a` , where `a` was not defined, the second clause is illegal.

## Stage 3: Generating an Execution Plan

When the validation succeeds, an execution plan is generated in Stage 3. Its data structure is stored in the `src/planner` directory, with the following logical structure.

![Generating an Execution Plan](https://user-images.githubusercontent.com/57335825/111280198-86c12580-8676-11eb-8226-d4e7f9a8c474.png)

### Query Execution Flow

The generated execution plan is a directed acyclic graph where the dependencies between nodes are determined in the [toPlan()](https://github.com/vesoft-inc/nebula-graph/blob/master/src/validator/PipeValidator.cpp) function of each module in the Validator. As shown in the preceding figure, the Project node depends on the Filter node, the Filter node depends on the GetNeighbor node, and so on, up to the Start node.

During the execution stage, the Executor generates an operator for each node and starts scheduling from the root node (e.g., the Project node in this example). If the root node is found to be dependent on other nodes, the Executor **recursively calls** the nodes that the root node depends on until it finds a node that is not dependent on any node (e.g., the Start node in this example), and then starts execution. After the execution, the Executor continues executing the node that depends on the executed node (e.g., the GetNeighbor node in this example), and so on, until the root node is reached.

### Query Data Flow

The input and output of each node are also determined in [toPlan()](https://github.com/vesoft-inc/nebula-graph/blob/master/src/validator/PipeValidator.cpp). Although the execution is performed in the order defined in the execution plan, the input of each node can be customized and does not completely depend on the previous node, because the input and output of all nodes are actually stored in a hash table, where the keys are the names customized during the creation of each node. For example, if the hash table is named as `ResultMap`, when creating the Filter node, you can determine that the node takes data from `ResultMap["GN1"]`, then puts the result into `ResultMap["Filter2"]`, and so on. All these work as the input and output of each node. The hash table is defined in `src/context/ExecutionContext.cpp` under the `nebula-graph` repository. The execution plan is not really executed, so the value of each key in the associated hash table is empty (except for the starting node, where the input variables hold the starting data), and they will be computed and filled in the Executor stage.

This is a simple example. A more complex one will be shown at the end for you to better understand how an execution plan works.

## Stage 4: Optimizing the Execution Plan

An execution plan generated in the preceding stage can be optimized optionally. In the `etc/nebula-graphd.conf` file, when `enable_optimizer` is set to `true`, the execution plans will be optimized. In this example, when optimization is enabled, the data flow is as follows.

![Optimizing the Execution Plan](https://user-images.githubusercontent.com/57335825/111280908-53cb6180-8677-11eb-83c5-01cb632309dd.png)

As shown in the preceding figure, the Filter node is integrated into the GetNeighbor node. Therefore, when the GetNeighbor operator calls interfaces of the Storage layer to get the neighboring edges of a vertex during the execution stage, the Storage layer will directly filter out the unqualified edges internally. Such optimization greatly reduces the amount of data transfer,which is commonly known as filter push down.

In the execution plan, each node directly depends on other nodes. To explore equivalent transformations and to reuse the same parts of an execution plan, such a direct dependency between nodes is converted into an OptGroupNode to OptGroup dependency. Each OptGroup contains an equivalent set of OptGroupNodes, and each OptGroupNode contains one node in the execution plan. An OptGroupNode depends on an OptGroup but not on other OptGroupNodes. Therefore, from one OptGroupNode, many equivalent execution plans can be explored because the OptGroup it depends on contains different OptGroupNodes, which saves storage space.

All the optimization rules we have implemented so far are considered as RBO (Rule-Based Optimization), which means the plans optimized based on the rules must be better than their original versions. The CBO (Cost-Based Optimization) feature is under development. The optimization process is a "bottom-up" exploration process: For each rule, the root node of the execution plan (in this case, the Project node) is the entry point, along the node dependencies, the bottom node is found, and then from the bottom node, the OptGroupNode in each OptGroup is explored to see if it matches the rule, until the entire execution plan can no longer apply the rule. Then, the next rule is explored.

This figure shows the optimization process of this example.

![Execution plan optimization process](https://user-images.githubusercontent.com/57335825/111280406-bf60ff00-8676-11eb-9f87-71390f706bac.png)

As shown in the preceding figure, when the Filter node is explored, it is found that its children node is GetNeighbors, which matches successfully with the pre-defined pattern in the rule, so a transformation is initiated to integrate the Filter node into the GetNeighbors node, the Filter node is removed, and then the process continues to the next rule.

The optimized code is in the `src/optimizer/` directory under the `nebula-graph` repository.

## Stage 5: Execution

In the fifth stage, the Scheduler generates the corresponding execution operators against the execution plan, starting from the leaf nodes and ending at the root node. The structure is as follows.

![Execution plan execution](https://user-images.githubusercontent.com/57335825/111280526-ea4b5300-8676-11eb-84fa-25353a0b86ed.png)

Each node of the execution plan has one execution operator node, whose input and output have been determined in the execution plan. Each operator only needs to get the values for the input variables, compute them, and finally put the results into the corresponding output variables. Therefore, it is only necessary to execute step by step from the starting node, and the result of the last operator is returned to the user as the final result.

# An Example Query

Now we will get acquainted with the structure of an execution plan by executing a `FIND SHORTEST PATH` statement.

Open nebula-console and enter the following statement.

```plain
FIND SHORTEST PATH FROM "YAO MING" TO "Tim Duncan" OVER like, serve UPTO 5 STEPS
```

To get the details of the execution plan of this statement, precede it with the`EXPLAIN`.

![Detailed execution plan](https://user-images.githubusercontent.com/57335825/111280632-07802180-8677-11eb-8c9d-e016bfee3130.png)

The preceding figure shows, from left to right, the unique ID of each node in the execution plan, their names, the IDs of their dependent nodes, the profiling data (information about the execution of the `PROFILE` command), and the details of each node (including the names of the input and output variables, the column names of the output results, and the parameters of the node).

To visualize the information, you can follow these steps:

1. Use `EXPLAIN format="dot"` instead of `EXPLAIN`. nebula-console can generate the data in the DOT format.

2. Open the [Graphviz Online](https://dreampuf.github.io/GraphvizOnline/) website and paste the generated DOT data. You can see the structure as shown in the following figure. This structure corresponds to the execution flow of the operators in the execution stage.

![Visualized execution plan](https://user-images.githubusercontent.com/57335825/111280728-1ff03c00-8677-11eb-84c8-5d67b6aa97ce.png)

The shortest path algorithm uses the two-way breadth-first search algorithm, expanding from both "YAO MING" and "Tim Duncan", so GetNeighbors, BFSShortest, Project, and Dedup have two operators, respectively, the input is connected by the PassThrough operator, and the path is stitched by the ConjunctPath operator. The LOOP operator then controls the number of steps to be extended outward, and it can be seen that the input of the DataCollect operator is actually taken from the output variable of the ConjunctPath operator.

The information of each operator is in the `src/executor` directory under the `nebula-graph` directory. Feel free to share your ideas with us on GitHub: [https://github.com/vesoft-inc/nebula](https://github.com/vesoft-inc/nebula) or the : [https://github.com/vesoft-inc/nebula/discussions](https://github.com/vesoft-inc/nebula/discussions)

>Hi, I’m Ming Quan, engineer at NebulaGraph. I’m interested in database storage and would like to share my experiences in this regard. Hope my post is of help to you. Please let me know if you have any ideas about this. Thanks.

This article introduces the overall structure of the Query layer and uses an nGQL statement to describe how it is processed in the four main modules of the Query layer. 