bitflicker64 commented on code in PR #464: URL: https://github.com/apache/hugegraph-doc/pull/464#discussion_r4006510533
########## content/cn/docs/quickstart/toolchain/hugegraph-seatunnel-connector.md: ########## @@ -0,0 +1,442 @@ +--- +title: "使用 SeaTunnel 导入与迁移图数据" +linkTitle: "SeaTunnel 数据集成" +weight: 5 +--- + +SeaTunnel 可以把数据库、Kafka 等数据源接入 HugeGraph，也可以在两张 HugeGraph 图之间迁移顶点和边。连接器分为两部分：**Source 负责读取，Sink 负责写入**，中间可以接 SeaTunnel 的数据转换组件。 + +> **版本要求：本文面向 SeaTunnel 3.0+（dev 分支）**。所有示例使用 `mappings`。SeaTunnel **2.3.13 只有 HugeGraph Sink**，使用旧的 `schema_config`，不能直接运行本文配置。 + +[](/cn/docs/images/seatunnel/seatunnel-vs-loader-en.png) + +点击配图可查看原图。 + +## 1 与 Loader 的区别及版本要求 + +[HugeGraph-Loader](/cn/docs/quickstart/toolchain/hugegraph-loader/) 面向图数据导入，用输入源和图映射描述“哪些记录变成哪些顶点或边”。SeaTunnel 则把任务组织成 **Source → Transform → Sink**，适合复用已有的连接器、转换步骤和数据处理管道。 + +| 对比点 | HugeGraph-Loader | SeaTunnel | +| --- | --- | --- | +| 任务配置 | JSON 映射文件，描述输入源、顶点和边 | HOCON 作业文件，组合 Source、Transform 和 Sink | +| 适合的需求 | 直接把数据导入 HugeGraph，字段和值映射已能满足需求 | 把 HugeGraph 接入已有 SeaTunnel 管道，或复用其连接器与转换组件 | +| 运行准备 | 使用 Loader 工具；也提供 Spark Loader、Flink CDC 集成 | 准备匹配版本的引擎和连接器；本文使用 Zeta local 模式 | + +**不要只按数据源或批量/流式来选。** 两者都支持 JDBC、Kafka 和图数据，Loader 也有字段/值映射、Spark 与 Flink CDC 集成。直接导入图时可先用 Loader；如果已经有 SeaTunnel 作业，通常在原管道中接入 HugeGraph 更方便。 + +新建 SeaTunnel 任务使用 3.0+ 开发版本和 `mappings`。本文固定到提交 [`35b2716`](https://github.com/apache/seatunnel/commit/35b2716cde7d4c91a24fc618a8d9cae90e213db3) 核对，对应 `3.0.0-SNAPSHOT`。使用其他提交时，请重新核对连接器配置。 + +| 能力 | 本文固定版本 | 2.3.13 | +| --- | --- | --- | +| 写入 HugeGraph | Sink，使用 `mappings` | Sink，使用 `schema_config` | +| 读取 HugeGraph 顶点和边 | 支持 Source | 不支持 | +| 自动创建缺失的图模型 | `mappings` 默认支持 | 需提前创建 | + +如果暂时必须使用 2.3.13，请按 [2.3.13 Sink 文档](https://github.com/apache/seatunnel/blob/2.3.13/docs/zh/connectors/sink/HugeGraph.md) 配置，不要混用本文示例。 + +## 2 准备环境 + +### 2.1 获取 SeaTunnel 3.0+ + +准备 JDK 11，并设置 `JAVA_HOME`。从开发分支获取源码，按上游[开发环境文档](https://github.com/apache/seatunnel/blob/35b2716cde7d4c91a24fc618a8d9cae90e213db3/docs/zh/developer/setup.md)构建发行包： + +```bash +git clone --branch dev https://github.com/apache/seatunnel.git +cd seatunnel +# 复现本文配置时，固定到本次核对的提交 +git checkout 35b2716cde7d4c91a24fc618a8d9cae90e213db3 +./mvnw clean package -pl seatunnel-dist -am -Dmaven.test.skip=true +``` + +解压 `seatunnel-dist/target/` 中生成的二进制包，后续命令都在解压后的 SeaTunnel 安装目录执行。需要更新功能时，可以切换到其他提交；引擎与连接器插件应来自同一版本，避免混装 2.3.13 的 JAR。 + +本文使用 SeaTunnel 自带的 **Zeta 引擎和 local 模式**。确认安装目录的 `connectors/` 中包含 HugeGraph，以及所需的 JDBC 或 Kafka 连接器；如果自定义构建没有包含它们，需补齐同一次构建产出的插件。JDBC 示例还需要将 MySQL 驱动 JAR 放入 `lib/`，驱动类为 `com.mysql.cj.jdbc.Driver`。 + +### 2.2 准备 HugeGraph 和数据源 + +先启动 [HugeGraph Server](/cn/docs/quickstart/hugegraph/hugegraph-server/)，创建可用于测试的图。本文示例使用 `hugegraph` 图、`DEFAULT` 图空间，请按服务端实际配置修改；图空间名称区分大小写。启用了身份验证时，在 HugeGraph Source 和 Sink 中填写 `username`、`password`。 + +下面的图模型贯穿 JDBC 和 Kafka 示例。`mappings` 默认会创建缺失的 PropertyKey、VertexLabel 和 EdgeLabel；已有图模型必须与配置兼容。 + +| 图元素 | 名称与属性 | +| --- | --- | +| 属性 | `name` 为 Text，`age` 和 `since` 为 Int | +| 顶点 | `person`，主键为 `name`，属性为 `name`、`age` | +| 边 | `knows`，从 `person` 指向 `person`，属性为 `since` | + +所有示例中的 `mysql`、`kafka`、`hugegraph` 都是占位主机名，需替换为 **SeaTunnel 运行环境可访问的地址**。容器中的 `127.0.0.1` 指向容器自身；同一 Docker 网络可使用服务名。`host` 只填主机名或 IP，端口单独填写。 + +## 3 从关系库导入（sql2graph） + +用两个任务完成导入：先把 `person` 表写成顶点，再把 `knows` 表写成边。这样写边时，两个端点都已经存在。 + +[](/cn/docs/images/seatunnel/seatunnel-records-to-graph-en.png) + +### 3.1 导入顶点 + +在 MySQL 的 `demo` 数据库中准备示例数据，并让配置中的账号有读取权限： + +```sql +CREATE TABLE person ( + name VARCHAR(64) PRIMARY KEY, + age INT NOT NULL +); +INSERT INTO person VALUES ('marko', 29), ('vadas', 27); +``` + +保存为 `config/sql2graph-person.conf`，将数据库账号和密码替换为实际值： + +```hocon +env { + job.mode = "BATCH" +} + +source { + Jdbc { + url = "jdbc:mysql://mysql:3306/demo?useSSL=false&serverTimezone=UTC" + driver = "com.mysql.cj.jdbc.Driver" + username = "seatunnel" + password = "change_me" + query = "SELECT name, age FROM person ORDER BY name" + } +} + +sink { + HugeGraph { + host = "hugegraph" + port = 8080 + graph_name = "hugegraph" + graph_space = "DEFAULT" + batch_failure_fallback = false + mappings = [ + { + type = "VERTEX" + label = "person" + idStrategy = "PRIMARY_KEY" + idFields = ["name"] + properties = ["name", "age"] + } + ] + } +} +``` + +```bash +./bin/seatunnel.sh --config ./config/sql2graph-person.conf -m local +``` + +在 Hubble 或 Gremlin 中检查结果，应能查到 `marko` 和 `vadas` 及其年龄： + +```groovy +g.V().hasLabel('person').valueMap('name', 'age') +``` + +`idFields = ["name"]` 表示使用名字生成主键。重复导入同一个 `name` 会写到同一个顶点；`properties` 指定要写入的源字段。 + +### 3.2 导入边 + +准备关系表，其中两个端点字段对应前面导入的 `person.name`： + +```sql +CREATE TABLE knows ( + source_name VARCHAR(64) NOT NULL, + target_name VARCHAR(64) NOT NULL, + since INT NOT NULL +); +INSERT INTO knows VALUES ('marko', 'vadas', 2010); +``` + +<details> +<summary>展开配置，保存为 config/sql2graph-knows.conf</summary> + +```hocon +env { + job.mode = "BATCH" +} + +source { + Jdbc { + url = "jdbc:mysql://mysql:3306/demo?useSSL=false&serverTimezone=UTC" + driver = "com.mysql.cj.jdbc.Driver" + username = "seatunnel" + password = "change_me" + query = "SELECT source_name, target_name, since FROM knows ORDER BY source_name, target_name" + } +} + +sink { + HugeGraph { + host = "hugegraph" + port = 8080 + graph_name = "hugegraph" + graph_space = "DEFAULT" + batch_failure_fallback = false + check_vertex = true + mappings = [ + { + type = "EDGE" + label = "knows" + sourceConfig = { + label = "person" + idFields = ["source_name"] + } + targetConfig = { + label = "person" + idFields = ["target_name"] + } + fieldMapping = { + source_name = "name" + target_name = "name" + } + properties = ["since"] + } + ] + } +} +``` + +</details> + +确认顶点任务成功后，再执行边任务： + +```bash +./bin/seatunnel.sh --config ./config/sql2graph-knows.conf -m local +``` + +下面的查询应返回 `marko` 到 `vadas` 的 `knows` 边，属性 `since` 为 `2010`： + +```groovy +g.V().has('person', 'name', 'marko').outE('knows').where(inV().has('name', 'vadas')).valueMap() +``` + +`sourceConfig` 和 `targetConfig` 指定端点字段，`fieldMapping` 将它们对应到顶点主键 `name`，`properties = ["since"]` 只写边属性。示例启用 `check_vertex = true`，并关闭失败后逐条跳过的回退（`batch_failure_fallback = false`）；端点不存在或写入失败时，任务会报错。 + +如果关系表只有数字外键，而图的主键使用姓名，请先在 SQL 中关联出姓名，再交给 Sink。MySQL CDC 接入方式见 [MySQL CDC Source](https://seatunnel.apache.org/docs/connectors/source/MySQL-CDC/)。 + +## 4 从 Kafka 导入（kafka2graph） + +Kafka 适合持续接收事件。先创建 `user-events` topic，再写入以下 JSON 消息，每条消息对应一个 `person` 顶点： + +```json +{"name":"marko","age":29} +``` + +保存为 `config/kafka2graph.conf`： + +```hocon +env { + job.mode = "STREAMING" + checkpoint.interval = 10000 + sink.flush.interval = 5000 +} + +source { + Kafka { + bootstrap.servers = "kafka:9092" + topic = "user-events" + consumer.group = "hugegraph-import" + start_mode = "earliest" + format = "json" + schema = { + fields = { + name = "string" + age = "int" + } + } + } +} + +sink { + HugeGraph { + host = "hugegraph" + port = 8080 + graph_name = "hugegraph" + graph_space = "DEFAULT" + batch_failure_fallback = false + mappings = [ + { + type = "VERTEX" + label = "person" + idStrategy = "PRIMARY_KEY" + idFields = ["name"] + properties = ["name", "age"] + } + ] + } +} +``` + +```bash +./bin/seatunnel.sh --config ./config/kafka2graph.conf -m local +``` + +用第 3.1 节的 Gremlin 查询检查数据。流式任务会持续运行；`checkpoint.interval` 每 10 秒保存一次任务状态，`sink.flush.interval` 让 Zeta 每 5 秒触发一次刷新，避免少量消息一直等到批次填满。 + +HugeGraph Sink 是 **at-least-once（至少一次）** 写入，故障恢复可能重放记录。使用 `PRIMARY_KEY` 能让相同 `name` 落到同一个顶点，但不等于所有更新操作都具备 exactly-once 语义。定时刷新由 Zeta 提供，不适用于 Spark 或 Flink 引擎。 + +## 5 迁移 HugeGraph 图（graph2graph） + +下面从源图迁移 `person` 顶点和 `knows` 边。请使用独立的目标图：本节采用 `CUSTOMIZE_STRING` 保留顶点 ID，不要复用前面已经创建为 `PRIMARY_KEY` 的 `person` 标签。 + +这两个任务只迁移指定标签和属性，不会完整复制源图的索引、TTL 等全部 Schema 配置。运行期间应暂停源图写入，避免两个任务读到不同时间的数据；完成后核对顶点、边数量及抽样属性。 + +[](/cn/docs/images/seatunnel/seatunnel-preserve-ids-en.png) + +### 5.1 先迁移顶点 + +Source 自动补充 `~id` 保留列，Sink 把原 ID 作为字符串保存。无需在 `schema.fields` 中声明 `~id`，手动声明保留列会被拒绝。 + +<details> +<summary>展开配置，保存为 config/graph2graph-person.conf</summary> + +```hocon +env { + job.mode = "BATCH" +} + +source { + HugeGraph { + host = "source-hugegraph" + port = 8080 + graph_name = "hugegraph" + graph_space = "DEFAULT" + label = "person" + label_type = "VERTEX" + schema = { + fields = { + name = "string" + age = "int" + } + } + } +} + +sink { + HugeGraph { + host = "target-hugegraph" + port = 8080 + graph_name = "hugegraph" + graph_space = "DEFAULT" + batch_failure_fallback = false + mappings = [ + { + type = "VERTEX" + label = "person" + idStrategy = "CUSTOMIZE_STRING" + idFields = ["~id"] + properties = ["name", "age"] + } + ] + } +} +``` + +</details> + +```bash +./bin/seatunnel.sh --config ./config/graph2graph-person.conf -m local +``` + +### 5.2 再迁移边 + +确认顶点任务成功后，使用 Source 自动补充的 `~source_id` 和 `~target_id` 定位端点。因为上一任务保留了原 ID，这两列可以直接引用目标图中的顶点。 + +<details> +<summary>展开配置，保存为 config/graph2graph-knows.conf</summary> + +```hocon +env { + job.mode = "BATCH" +} + +source { + HugeGraph { + host = "source-hugegraph" + port = 8080 + graph_name = "hugegraph" + graph_space = "DEFAULT" + label = "knows" + label_type = "EDGE" + schema = { + fields = { + since = "int" + } + } + } +} + +sink { + HugeGraph { + host = "target-hugegraph" + port = 8080 + graph_name = "hugegraph" + graph_space = "DEFAULT" + check_vertex = true + batch_failure_fallback = false + mappings = [ + { + type = "EDGE" + label = "knows" + sourceConfig = { + label = "person" + idFields = ["~source_id"] + } + targetConfig = { + label = "person" + idFields = ["~target_id"] + } + properties = ["since"] + } + ] + } +} +``` + +</details> + +```bash +./bin/seatunnel.sh --config ./config/graph2graph-knows.conf -m local +``` + +本例打开端点检查，并让写入错误直接导致任务失败。默认的 `check_vertex = false` 不保证最终一致：缺少端点可能产生悬空边，因此不能用任务成功代替迁移结果检查。 + +> **为什么保留 ID？** HugeGraph 的 `PRIMARY_KEY` ID 包含顶点标签的内部 ID，两张图可能不同。例如源图顶点是 `1:marko`，目标图重新按主键生成的可能是 `2:marko`。如果重新生成顶点 ID 后仍复用源图的边端点，边就会连错。本例将原 ID 保存为字符串，因此会改变目标图的 ID 策略。 + +若要一次读取全部标签，省略 Source 的 `label` 后会按 `label_type` 为每个标签输出一张表；这时需用 `source_table` 将各 Sink 映射绑定到对应表，不能直接套用本节的单标签配置。其他限制见 [HugeGraph Source 文档](https://github.com/apache/seatunnel/blob/35b2716cde7d4c91a24fc618a8d9cae90e213db3/docs/zh/connectors/source/HugeGraph.md)。 Review Comment: ⚠️ important: at the pinned commit the Sink does not accept `source_table` inside `mappings`. `HugeGraphSinkOptions.MAPPINGS` is `listType(MappingConfig.class)`, and `ConfigUtil.convertValue` deserializes it with a plain Jackson `ObjectMapper`, which rejects unknown properties. `MappingConfig` is a Lombok `@Data` class with the field `sourceTable` and no Jackson annotations, so the accepted key is `sourceTable`, the same camelCase style as `idStrategy`/`idFields` in this guide. A mapping written as `source_table = "..."` fails at job startup with a JSON parsing error. The upstream error messages say `source_table`, but the parser does not accept that spelling. Please change the note to `sourceTable`, ideally with a one-line example such as `sourceTable = "<table path of that label>"`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
