Flink connector guide¶
The Flink connector 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.
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.
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());