Source: https://dev-docs.nebula-graph.io/v5.3.2/clients/flink-connector/
Documentation version: v5.3.2

# Flink connector guide

The [Flink connector](https://github.com/vesoft-inc/nebula-flink-connector/tree/v5.3.0) enables data transfer between NebulaGraph Database Developer Edition and Apache Flink. This topic introduces how to set up the Flink connector to read and write data from and to NebulaGraph Database Developer Edition.

## Prerequisites

- Java version 8 or later is installed.
- A NebulaGraph Database Developer Edition service group is running. For more information, see [Getting started with NebulaGraph Database Developer Edition](https://dev-docs.nebula-graph.io/v5.3.2/get-started/deploy/).

## Setup

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

````
<dependency>
    <groupId>com.vesoft</groupId>
    <artifactId>nebula-flink-connector</artifactId>
    <version>5.3.0</version>
</dependency>

````

## Common configuration

All examples in this topic share the following setup code. It is shown once here and omitted from subsequent examples for brevity.

Replace `<current-password>` with the current password for the specified account. If the password has changed, use the new password.

````
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

ConnectionOptions connectionOptions = ConnectionOptions
                .builder()
                .withGraphAddress("127.0.0.1:9669")
                .withUser("root")
                .withPassword("<current-password>")
                .build();

````

For the full list of `ConnectionOptions` parameters, see [Flink connector configurations](https://dev-docs.nebula-graph.io/v5.3.2/clients/flink-connector/flink-options/).

## Write data to NebulaGraph Database Developer Edition

### Write nodes

To write data from Flink into NebulaGraph Database Developer Edition as nodes of a specific node type:

````
// ... env and connectionOptions setup omitted

SinkNodeOptions sinkNodeOptions = SinkNodeOptions.builder()
                        .withGraphName("flinkSink")
                        .withNodeType("person")
                        .withFlinkFields(Arrays.asList("c0", "c1", "c2", "c3", "c4", "c5",
                                "c6", "c7", "c8", "c9", "c10", "c11", "c12"))
                        .withNebulaFields(Arrays.asList("col1", "col2", "col3", "col4", "col5",
                                "col6", "col7", "col8", "col9", "col10", "col11", "col12", "col13"))
                        .withWriteMode(WriteModeEnum.INSERTREPLACE)
                        .withBatchSize(2)
                        .build();

NebulaNodeBatchOutputFormat outputFormat       = new NebulaNodeBatchOutputFormat(connectionOptions, sinkNodeOptions);
NebulaSinkFunction<Row>     nebulaSinkFunction = new NebulaSinkFunction<>(outputFormat);
DataStream<Row> dataStream = playerSource.map(row -> {
            org.apache.flink.types.Row record = Row.withNames();
            for (int i = 0; i < row.size(); i++) {
                record.setField("c" + i, row.get(i));
            }
            return record;
        });
dataStream.addSink(nebulaSinkFunction);
env.execute("Write Nebula");

````

### Write edges

To write data from Flink into NebulaGraph Database Developer Edition as edges of a specific edge type:

````
// ... env and connectionOptions setup omitted

SinkEdgeOptions sinkEdgeOptions = SinkEdgeOptions.builder()
                        .withGraphName("flinkSink")
                        .withEdgeType("friend")
                        .withFlinkSrcPkFields(Arrays.asList("c0"))
                        .withNebulaSrcPks(Arrays.asList("col1"))
                        .withFlinkDstPkFields(Arrays.asList("c1"))
                        .withNebulaDstPks(Arrays.asList("col1"))
                        .withFlinkFields(Arrays.asList("c2", "c3", "c4", "c5", "c6", "c7",
                                "c8", "c9", "c10", "c11", "c12", "c13", "c14"))
                        .withNebulaFields(Arrays.asList("col1", "col2", "col3", "col4", "col5",
                                "col6", "col7", "col8", "col9", "col10", "col11", "col12", "col13"))
                        .withWriteMode(WriteModeEnum.INSERTIGNORE)
                        .withBatchSize(2)
                        .build();

NebulaEdgeBatchOutputFormat outputFormat = new NebulaEdgeBatchOutputFormat(connectionOptions, sinkEdgeOptions);
NebulaSinkFunction<Row> nebulaSinkFunction = new NebulaSinkFunction<>(outputFormat);
DataStream<Row> dataStream = friendSource.map(row -> {
            org.apache.flink.types.Row record = Row.withNames();
            for (int i = 0; i < row.size(); i++) {
                record.setField("c" + i, row.get(i));
            }
            return record;
        });
dataStream.addSink(nebulaSinkFunction);
env.execute("Write Nebula Edge");

````

### Write using custom GQL templates

By default, the connector automatically generates GQL statements based on the configured options. For advanced scenarios, you can use `withGqlTemplate()` to provide a custom GQL template that gives you full control over the generated statement.

Tip

Use the standard options for simple field-to-property mapping — they are high-performance and require minimal configuration. Use GQL templates when you need complex logic such as `CAST` type conversions, calling UDFs, or custom `MATCH` logic before insertion.

A custom GQL template is a string containing placeholders that the connector replaces at runtime. The `{{ TABLE }}` placeholder is required — the connector substitutes it with the `TABLE` clause built from the current batch of Flink data.

The generated `TABLE` clause always uses `t` as the table name. You can reference the table columns in subsequent statements using `r.c0`, `r.c1`, `r.c2`, ..., where the number is the zero-based index of the Flink row column. For edges, source/destination primary key columns are named `r.src_0`, `r.dst_0`, etc.

The template can contain any valid GQL statement after `{{ TABLE }}`. For example, a minimal template that ignores the table data and inserts fixed values:

````
{{ TABLE }} USE movie INSERT (:Movie{id:10000,name:"movie"})

````

Available placeholders

 | Placeholder | Description
 | `{{ TABLE }}` | Required. The `TABLE` clause for batch data. The connector automatically builds this from the Flink data stream.
 | `{{ GRAPH }}` | The graph name, i.e., the value of `withGraphName()`.
 | `{{ TYPE }}` | The node type or edge type name.

#### Node example

The following example uses `CAST` to explicitly convert data types during insertion:

````
// ... env and connectionOptions setup omitted

String gqlTemplate = "{{ TABLE }} \n"
        + "USE `{{ GRAPH }}` \n"
        + "FOR r IN t \n"
        + "INSERT OR REPLACE (@`{{ TYPE }}`{"
        + "`col1`:r.c0,`col2`:r.c1,`col3`:CAST(r.c2 AS INT8),`col4`:CAST(r.c3 AS INT16),"
        + "`col5`:r.c4,`col6`:r.c5,`col7`:r.c6,`col8`:r.c7,`col9`:r.c8,`col10`:r.c9,"
        + "`col11`:r.c10,`col12`:r.c11,`col13`:r.c12})";

SinkNodeOptions sinkNodeOptions = SinkNodeOptions.builder()
                        .withGraphName("flinkSink")
                        .withNodeType("person")
                        .withFlinkFields(Arrays.asList("c0", "c1", "c2", "c3", "c4", "c5",
                                "c6", "c7", "c8", "c9", "c10", "c11", "c12"))
                        .withNebulaFields(Arrays.asList("col1", "col2", "col3", "col4", "col5",
                                "col6", "col7", "col8", "col9", "col10", "col11", "col12", "col13"))
                        .withWriteMode(WriteModeEnum.INSERTREPLACE)
                        .withGqlTemplate(gqlTemplate)   // Use custom GQL template
                        .withBatchSize(10)
                        .build();

NebulaNodeBatchOutputFormat outputFormat       = new NebulaNodeBatchOutputFormat(connectionOptions, sinkNodeOptions);
NebulaSinkFunction<Row>     nebulaSinkFunction = new NebulaSinkFunction<>(outputFormat);
DataStream<Row> dataStream = playerSource.map(row -> {
            org.apache.flink.types.Row record = Row.withNames();
            for (int i = 0; i < row.size(); i++) {
                record.setField("c" + i, row.get(i));
            }
            return record;
        });
dataStream.addSink(nebulaSinkFunction);
env.execute("Write Nebula with Template GQL");

````

#### Edge example

For edges, the `{{ TABLE }}` clause includes columns for source/destination node primary keys (`src_0`, `dst_0`, ...) and edge properties (`c0`, `c1`, ...):

````
// ... env and connectionOptions setup omitted

String gqlTemplate = "{{ TABLE }}\n"
        + "USE `{{ GRAPH }}` \n"
        + "FOR r IN t \n"
        + "OPTIONAL MATCH (n_src@`person`) WHERE n_src.`col1`=r.src_0 \n"
        + "OPTIONAL MATCH (n_dst@`person`) WHERE n_dst.`col1`=r.dst_0 \n"
        + "INSERT OR REPLACE (n_src)-[@`{{ TYPE }}`{"
        + "`col1`:r.c0,`col2`:r.c1,`col3`:CAST(r.c2 AS INT8),`col4`:CAST(r.c3 AS INT16),"
        + "`col5`:r.c4,`col6`:r.c5,`col7`:r.c6,`col8`:r.c7,`col9`:r.c8,`col10`:r.c9,"
        + "`col11`:r.c10,`col12`:r.c11,`col13`:r.c12}]->(n_dst)";

SinkEdgeOptions sinkEdgeOptions = SinkEdgeOptions.builder()
                        .withGraphName("flinkSink")
                        .withEdgeType("friend")
                        .withFlinkSrcPkFields(Arrays.asList("c0"))
                        .withNebulaSrcPks(Arrays.asList("col1"))
                        .withFlinkDstPkFields(Arrays.asList("c1"))
                        .withNebulaDstPks(Arrays.asList("col1"))
                        .withFlinkFields(Arrays.asList("c2", "c3", "c4", "c5", "c6", "c7",
                                "c8", "c9", "c10", "c11", "c12", "c13", "c14"))
                        .withNebulaFields(Arrays.asList("col1", "col2", "col3", "col4", "col5",
                                "col6", "col7", "col8", "col9", "col10", "col11", "col12", "col13"))
                        .withWriteMode(WriteModeEnum.INSERTREPLACE)
                        .withGqlTemplate(gqlTemplate)   // Use custom GQL template
                        .withBatchSize(10)
                        .build();

NebulaEdgeBatchOutputFormat outputFormat = new NebulaEdgeBatchOutputFormat(connectionOptions, sinkEdgeOptions);
NebulaSinkFunction<Row> nebulaSinkFunction = new NebulaSinkFunction<>(outputFormat);
DataStream<Row> dataStream = friendSource.map(row -> {
            org.apache.flink.types.Row record = Row.withNames();
            for (int i = 0; i < row.size(); i++) {
                record.setField("c" + i, row.get(i));
            }
            return record;
        });
dataStream.addSink(nebulaSinkFunction);
env.execute("Write Nebula Edge with Template GQL");

````

## Read data from NebulaGraph Database Developer Edition

### Read nodes

To read nodes of a specific node type from NebulaGraph Database Developer Edition and load them into Flink:

````
// ... env and connectionOptions setup omitted

SourceExecutionOptions nodeExecutionOptions = SourceNodeOptions.builder()
                .withGraphName("flinkSource")
                .withNodeType("person")
                .withReturnCols(null)
                .withBatchSize(10)
                .build();

NebulaInputRowFormat inputRowFormat = new NebulaInputRowFormat(connectionOptions, nodeExecutionOptions);
DataSource<Row> rowDataSource = env.createInput(inputRowFormat);

System.out.println("rowDataSource count: " + rowDataSource.count());

````

### Read edges

To read edges of a specific edge type from NebulaGraph Database Developer Edition and load them into Flink:

````
// ... env and connectionOptions setup omitted

SourceExecutionOptions edgeExecutionOptions = SourceEdgeOptions.builder()
                .withGraphName("flinkSource")
                .withEdgeType("friend")
                .withReturnCols(null)
                .withBatchSize(10)
                .build();

NebulaInputRowFormat inputRowFormat = new NebulaInputRowFormat(connectionOptions, edgeExecutionOptions);
DataSource<Row> rowDataSource = env.createInput(inputRowFormat);

System.out.println("rowDataSource count: " + rowDataSource.count());

````

## More information

- [Flink connector GitHub repository](https://github.com/vesoft-inc/nebula-flink-connector/tree/v5.3.0)
- [Flink connector configurations](https://dev-docs.nebula-graph.io/v5.3.2/clients/flink-connector/flink-options/)