Source: https://dev-docs.yueshu.com.cn/v5.3.2/clients/flink-connector/ Documentation version: v5.3.2 # Flink 连接器指南 [Flink 连接器](https://github.com/vesoft-inc/nebula-flink-connector/tree/v5.3.0)用于实现悦数图数据库开发版与 Apache Flink 间数据传输。本文介绍如何设置 Flink 连接器以实现数据传输。 ## 前提条件 - 已安装 Java 8 或更高版本。 - 悦数图数据库开发版服务组正在运行中。更多信息，请参见[快速开始](https://dev-docs.yueshu.com.cn/v5.3.2/get-started/deploy/)。 ## 安装将 Flink 连接器导入到您的 Flink 项目中。例如，使用 Maven 导入时，需将以下依赖项添加到`pom.xml`文件： ```` com.vesoft nebula-flink-connector 5.3.0 ```` ## 通用配置本文所有示例共享以下初始化代码。此处展示一次，后续示例中省略。将``替换为登录账号的当前密码。若已修改密码，请使用修改后的密码。 ```` StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); ConnectionOptions connectionOptions = ConnectionOptions .builder() .withGraphAddress("127.0.0.1:9669") .withUser("root") .withPassword("") .build(); ```` 有关`ConnectionOptions`的完整参数说明，请参见 [Flink 连接器配置](https://dev-docs.yueshu.com.cn/v5.3.2/clients/flink-connector/flink-options/)。 ## 向悦数图数据库开发版写入数据 ### 写入点将 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 nebulaSinkFunction = new NebulaSinkFunction<>(outputFormat); DataStream 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 nebulaSinkFunction = new NebulaSinkFunction<>(outputFormat); DataStream 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 nebulaSinkFunction = new NebulaSinkFunction<>(outputFormat); DataStream 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 nebulaSinkFunction = new NebulaSinkFunction<>(outputFormat); DataStream 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 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 rowDataSource = env.createInput(inputRowFormat); System.out.println("rowDataSource count: " + rowDataSource.count()); ```` ## 更多信息 - [Flink 连接器 GitHub 仓库](https://github.com/vesoft-inc/nebula-flink-connector/tree/v5.3.0) - [Flink 连接器配置](https://dev-docs.yueshu.com.cn/v5.3.2/clients/flink-connector/flink-options/)