Flink 连接器指南¶

Flink 连接器用于实现悦数图数据库开发版与 Apache Flink 间数据传输。本文介绍如何设置 Flink 连接器以实现数据传输。

前提条件¶

已安装 Java 8 或更高版本。
悦数图数据库开发版服务组正在运行中。更多信息，请参见快速开始。

安装¶

将 Flink 连接器导入到您的 Flink 项目中。例如，使用 Maven 导入时，需将以下依赖项添加到pom.xml文件：

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

通用配置¶

本文所有示例共享以下初始化代码。此处展示一次，后续示例中省略。

将<current-password>替换为登录账号的当前密码。若已修改密码，请使用修改后的密码。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

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

有关ConnectionOptions的完整参数说明，请参见 Flink 连接器配置。

向悦数图数据库开发版写入数据¶

写入点¶

将 Flink 中的数据写入悦数图数据库开发版作为特定点类型的点：

// ... env 和 connectionOptions 初始化代码省略

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");

写入边¶

将 Flink 中的数据写入悦数图数据库开发版作为特定边类型的边：

// ... env 和 connectionOptions 初始化代码省略

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");

使用自定义 GQL 模板写入¶

默认情况下，连接器会根据配置选项自动生成 GQL 语句。对于进阶场景，可以使用withGqlTemplate()提供自定义 GQL 模板，从而完全控制生成的语句。

Tip

对于简单的字段到属性映射，推荐使用标准选项——它们性能更高，配置更简单。当需要复杂逻辑（如CAST类型转换、调用 UDF、或在插入前自定义MATCH逻辑）时，使用 GQL 模板。

自定义 GQL 模板是一个包含占位符的字符串，连接器会在运行时替换这些占位符。{{ TABLE }}占位符必填——连接器会将其替换为由当前批次 Flink 数据构建的TABLE子句。

生成的TABLE子句的表名固定为t。在后续语句中，可以通过r.c0、r.c1、r.c2等引用表中的列，其中数字为 Flink 数据行中列的从零开始的索引。对于边，起始/目标点主键列的命名为r.src_0、r.dst_0等。

模板中{{ TABLE }}之后可以是任意合法的 GQL 语句。例如，一个不引用表数据而插入固定值的最简模板：

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

可用占位符

占位符	描述
`{{ TABLE }}`	必填。批量数据的`TABLE`子句。连接器自动将 Flink 数据流中的数据构建为 TABLE 子句。
`{{ GRAPH }}`	图名称，即`withGraphName()`的值。
`{{ TYPE }}`	点类型或边类型名称。

点示例¶

以下示例使用CAST在插入时显式转换数据类型：

// ... env 和 connectionOptions 初始化代码省略

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)   // 使用自定义 GQL 模板
                        .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");

边示例¶

对于边，{{ TABLE }}子句包含起始/目标点主键列（src_0、dst_0等）和边属性列（c0、c1等）：

// ... env 和 connectionOptions 初始化代码省略

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)   // 使用自定义 GQL 模板
                        .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");

从悦数图数据库开发版读取数据¶

读取点¶

读取悦数图数据库开发版中特定点类型的点并将其加载到 Flink 中：

// ... env 和 connectionOptions 初始化代码省略

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

读取边¶

读取悦数图数据库开发版中特定边类型的边并将其加载到 Flink 中：

// ... env 和 connectionOptions 初始化代码省略

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