imbajin commented on code in PR #464: URL: https://github.com/apache/hugegraph-doc/pull/464#discussion_r4018074777
########## 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` 为每个标签输出一张表；这时需用 `sourceTable` 将各 Sink 映射绑定到对应表，例如 `sourceTable = "person"`。不能直接套用本节的单标签配置。其他限制见 [HugeGraph Source 文档](https://github.com/apache/seatunnel/blob/35b2716cde7d4c91a24fc618a8d9cae90e213db3/docs/zh/connectors/source/HugeGraph.md)。 Review Comment: Fixed in e63c184d2. The Chinese and English guides now document one output table per label and use the full table path, with `sourceTable = "default.person"` as the example. They also tell readers to use the exact table name shown in the Writer log. ########## content/en/docs/quickstart/toolchain/hugegraph-seatunnel-connector.md: ########## @@ -0,0 +1,442 @@ +--- +title: "Import and Migrate Graph Data with SeaTunnel" +linkTitle: "SeaTunnel Data Integration" +weight: 5 +--- + +SeaTunnel connects data sources such as databases and Kafka to HugeGraph. It can also migrate vertices and edges between two HugeGraph graphs. The connector has two parts: **Source reads data and Sink writes data**, with SeaTunnel transform components available between them. + +> **Version requirement: This guide targets SeaTunnel 3.0+ (dev branch).** All examples use `mappings`. SeaTunnel **2.3.13 includes only the HugeGraph Sink**, which uses the legacy `schema_config` and cannot run the configurations in this guide. + +[](/docs/images/seatunnel/seatunnel-vs-loader-en.png) + +Click a diagram to view the original size. + +## 1 Loader vs SeaTunnel and version requirements + +[HugeGraph-Loader](/docs/quickstart/toolchain/hugegraph-loader/) is designed for graph data imports. Its input sources and graph mappings describe which records become vertices or edges. SeaTunnel organizes a job as **Source → Transform → Sink**, so you can reuse existing connectors, transforms, and data pipelines. + +| Comparison | HugeGraph-Loader | SeaTunnel | +| --- | --- | --- | +| Job configuration | JSON mapping file describing the source, vertices, and edges | HOCON job file combining Source, Transform, and Sink | +| Best for | Directly importing data into HugeGraph when field and value mappings are enough | Connecting HugeGraph to an existing SeaTunnel pipeline or reusing its connectors and transforms | +| Runtime setup | Use Loader; Spark Loader and Flink CDC integrations are also available | Prepare matching engine and connector versions; this guide uses Zeta in local mode | + +**Choose by the work you need to complete, not only by the data source or batch/stream mode.** Both tools support JDBC, Kafka, and graph data. Loader also provides field/value mappings and Spark or Flink CDC integrations. Use Loader for a direct graph import; if a SeaTunnel job already exists, adding HugeGraph to that pipeline is usually simpler. + +For new SeaTunnel jobs, use the 3.0+ development version and `mappings`. This guide was checked against commit [`35b2716`](https://github.com/apache/seatunnel/commit/35b2716cde7d4c91a24fc618a8d9cae90e213db3), which corresponds to `3.0.0-SNAPSHOT`. Recheck the connector configuration when using another commit. + +| Capability | Version used by this guide | 2.3.13 | +| --- | --- | --- | +| Write to HugeGraph | Sink with `mappings` | Sink with `schema_config` | +| Read HugeGraph vertices and edges | Source supported | Not supported | +| Create missing graph schema automatically | Enabled by default through `mappings` | Create the schema first | + +If you must use 2.3.13, follow the [2.3.13 Sink documentation](https://github.com/apache/seatunnel/blob/2.3.13/docs/en/connectors/sink/HugeGraph.md) and do not mix it with the examples in this guide. + +## 2 Prepare the environment + +### 2.1 Get SeaTunnel 3.0+ + +Install JDK 11 and set `JAVA_HOME`. Clone the development branch and build a distribution by following the upstream [development setup guide](https://github.com/apache/seatunnel/blob/35b2716cde7d4c91a24fc618a8d9cae90e213db3/docs/en/developer/setup.md): + +```bash +git clone --branch dev https://github.com/apache/seatunnel.git +cd seatunnel +# Pin to the checked commit when reproducing this guide +git checkout 35b2716cde7d4c91a24fc618a8d9cae90e213db3 +./mvnw clean package -pl seatunnel-dist -am -Dmaven.test.skip=true +``` + +Extract the binary package from `seatunnel-dist/target/`. Run the remaining commands from the extracted SeaTunnel installation directory. When updating the feature set, switch to another commit as needed. Keep the engine and connector plugins from the same build, and do not mix in 2.3.13 JARs. + +This guide uses the bundled **Zeta engine in local mode**. Check that `connectors/` contains HugeGraph and the JDBC or Kafka connector required by each example. If a custom build does not include them, add the plugins produced by that same build. The JDBC examples also require the MySQL driver JAR in `lib/`, with driver class `com.mysql.cj.jdbc.Driver`. + +### 2.2 Prepare HugeGraph and data sources + +Start [HugeGraph Server](/docs/quickstart/hugegraph/hugegraph-server/) and create a graph for testing. The examples use the `hugegraph` graph in the `DEFAULT` graph space. Adjust these names to match the server configuration; graph space names are case-sensitive. If authentication is enabled, provide `username` and `password` in the HugeGraph Source and Sink configurations. + +The following graph model is shared by the JDBC and Kafka examples. `mappings` creates missing PropertyKey, VertexLabel, and EdgeLabel definitions by default; existing schema definitions must be compatible. + +| Graph element | Name and properties | +| --- | --- | +| Properties | `name` is Text; `age` and `since` are Int | +| Vertex | `person`, primary key `name`, properties `name` and `age` | +| Edge | `knows`, from `person` to `person`, property `since` | + +The `mysql`, `kafka`, and `hugegraph` host names in the examples are placeholders. Replace them with addresses reachable from the SeaTunnel runtime. Inside a container, `127.0.0.1` points to that container; services on the same Docker network can use their service names. Set `host` to a host name or IP address, and set the port separately. + +## 3 Import from a relational database (sql2graph) + +Use two jobs for this import: write the `person` table as vertices first, then write the `knows` table as edges. Both edge endpoints will already exist when the edge job runs. + +[](/docs/images/seatunnel/seatunnel-records-to-graph-en.png) + +### 3.1 Import vertices + +Prepare the sample data in the MySQL `demo` database and grant the configured account read access: + +```sql +CREATE TABLE person ( + name VARCHAR(64) PRIMARY KEY, + age INT NOT NULL +); +INSERT INTO person VALUES ('marko', 29), ('vadas', 27); +``` + +Save the following as `config/sql2graph-person.conf` and replace the database user name and password: + +```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 +``` + +Check the result in Hubble or Gremlin. You should find `marko` and `vadas` with their ages: + +```groovy +g.V().hasLabel('person').valueMap('name', 'age') +``` + +`idFields = ["name"]` uses the name to generate the primary key. Importing the same `name` again writes to the same vertex. `properties` lists the source fields to write. + +### 3.2 Import edges + +Prepare the relation table. Its two endpoint fields correspond to `person.name` from the vertex job: + +```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>Expand the configuration and save it as 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> + +After the vertex job succeeds, run the edge job: + +```bash +./bin/seatunnel.sh --config ./config/sql2graph-knows.conf -m local +``` + +The following query should return a `knows` edge from `marko` to `vadas` with `since` set to `2010`: + +```groovy +g.V().has('person', 'name', 'marko').outE('knows').where(inV().has('name', 'vadas')).valueMap() +``` + +`sourceConfig` and `targetConfig` identify the endpoint fields. `fieldMapping` maps them to the vertex primary key `name`, and `properties = ["since"]` writes only the edge property. The example enables `check_vertex = true` and disables per-record fallback after a batch failure (`batch_failure_fallback = false`), so a missing endpoint or write failure causes the job to fail. + +If the relation table has only numeric foreign keys while the graph uses names as primary keys, join the names in SQL before passing the records to the Sink. See [MySQL CDC Source](https://seatunnel.apache.org/docs/connectors/source/MySQL-CDC/) for MySQL CDC integration. + +## 4 Import from Kafka (kafka2graph) + +Kafka is useful for a continuous stream of events. Create the `user-events` topic and publish the following JSON message. Each message becomes one `person` vertex: + +```json +{"name":"marko","age":29} +``` + +Save the following as `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 +``` + +Use the Gremlin query from section 3.1 to check the data. The streaming job keeps running. `checkpoint.interval` saves job state every 10 seconds, while `sink.flush.interval` asks Zeta to flush every 5 seconds so a small number of messages does not wait for a full batch. + +HugeGraph Sink writes with **at-least-once** semantics, so recovery can replay records. `PRIMARY_KEY` sends the same `name` to the same vertex, but it does not make every update exactly-once. Scheduled flushing is provided by Zeta and does not apply to Spark or Flink engines. + +## 5 Migrate a HugeGraph graph (graph2graph) + +The following example migrates `person` vertices and `knows` edges from a source graph. Use a separate target graph. This section uses `CUSTOMIZE_STRING` to preserve vertex IDs. Do not reuse the `person` label created earlier with `PRIMARY_KEY`. + +These two jobs migrate only the selected labels and properties. They do not copy every source schema setting, such as indexes and TTLs. Pause writes to the source graph during the migration so both jobs read a consistent point in time. Afterward, compare vertex and edge counts and sample properties. + +[](/docs/images/seatunnel/seatunnel-preserve-ids-en.png) + +### 5.1 Migrate vertices first + +Source adds a `~id` column for the original ID, and Sink stores it as a string. Do not declare `~id` in `schema.fields`; manually declaring this reserved column is rejected. + +<details> +<summary>Expand the configuration and save it as 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 Migrate edges second + +After the vertex job succeeds, use the `~source_id` and `~target_id` columns added by Source to locate endpoints. Because the previous job preserved the original IDs, these columns can refer directly to vertices in the target graph. + +<details> +<summary>Expand the configuration and save it as 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 +``` + +This example checks endpoints and makes write errors fail the job. The default `check_vertex = false` does not guarantee a consistent result: a missing endpoint can create a dangling edge, so a successful job is not a substitute for checking the migrated graph. + +> **Why preserve IDs?** A HugeGraph `PRIMARY_KEY` ID contains the internal ID of the vertex label, and that internal ID can differ between graphs. For example, a source vertex can be `1:marko`, while regenerating the primary key in the target graph can produce `2:marko`. Reusing the source edge endpoints after regenerating vertex IDs can connect edges to the wrong vertices. This example stores the original ID as a string, which changes the target graph's ID strategy. + +When Source reads every label, omit `label` and it outputs one table per `label_type`. In that case, bind each Sink mapping to its table with `sourceTable`, for example `sourceTable = "person"`. Do not reuse the single-label configuration from this section. See the [HugeGraph Source documentation](https://github.com/apache/seatunnel/blob/35b2716cde7d4c91a24fc618a8d9cae90e213db3/docs/en/connectors/source/HugeGraph.md) for other limitations. Review Comment: Fixed in e63c184d2. The Chinese and English guides now document one output table per label and use the full table path, with `sourceTable = "default.person"` as the example. They also tell readers to use the exact table name shown in the Writer log. -- 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]
