This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-11953-af0a647d27bcc97b1e185d9a1292729622973df4 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit a0f5a56b86adbdcf2ab83ac7c96b058c241671d2 Author: lce_mo <[email protected]> AuthorDate: Sun Sep 6 03:12:11 2026 +0000 [Docs][Examples] Add batch MySQL to HDFS partitioned Parquet example (#11953) --- docs/en/getting-started/recipes/mysql-to-hdfs.md | 215 +++++++++++++++++++++++ docs/en/getting-started/recipes/overview.md | 1 + docs/sidebars.js | 1 + docs/zh/getting-started/recipes/mysql-to-hdfs.md | 215 +++++++++++++++++++++++ docs/zh/getting-started/recipes/overview.md | 1 + 5 files changed, 433 insertions(+) diff --git a/docs/en/getting-started/recipes/mysql-to-hdfs.md b/docs/en/getting-started/recipes/mysql-to-hdfs.md new file mode 100644 index 0000000000..cc0bded55b --- /dev/null +++ b/docs/en/getting-started/recipes/mysql-to-hdfs.md @@ -0,0 +1,215 @@ +--- +title: MySQL to HDFS +--- + +# MySQL to HDFS + +Use this recipe to batch-load MySQL orders into HDFS as Snappy-compressed Parquet files, partitioned by date. The pipeline uses a JDBC source, a SQL transform, and an HdfsFile sink. + +The transform renames `id` to `order_id`, normalizes order status to uppercase, filters negative amounts, and derives the `pt_dt` partition key. This is a batch snapshot example, not CDC or an incremental synchronization job. + +## Prerequisites + +1. Finish [Run your first job](../locally/run-your-first-job.md). This recipe uses SeaTunnel Zeta in local mode on Linux. Set `SEATUNNEL_HOME` to your extracted SeaTunnel distribution directory; a source checkout is not required. +2. Install `connector-jdbc` and `connector-file-hadoop` for the same SeaTunnel version as the distribution. Follow [Deployment](../locally/deployment.md) and include the following entries in `config/plugin_config`. Preserve any other connectors your environment needs. + +```plugin_config +--seatunnel-connectors-- +connector-jdbc +connector-file-hadoop +--end-- +``` + +3. Put the MySQL JDBC driver, such as `mysql-connector-j-8.x.jar`, in `${SEATUNNEL_HOME}/lib`. Install the connectors and check that they and the driver are present: + +```bash +cd "${SEATUNNEL_HOME}" +sh bin/install-plugin.sh +ls connectors | grep -E 'connector-(jdbc|file-hadoop)' +ls lib | grep 'mysql-connector' +``` + +The Zeta distribution includes Hadoop jars; inspect `lib` before adding dependencies. Do not mix arbitrary Hadoop client versions. See [HdfsFile sink](../../connectors/sink/HdfsFile.md) for environment-specific requirements. + +4. Prepare an accessible MySQL instance and an account with `SELECT` permission on the source table. Use a setup account with database/table creation and insert permissions for the seed SQL below; the job account does not need these setup permissions. +5. Prepare a reachable HDFS cluster and an unused output directory. The SeaTunnel process must have permission to write both the output and the sink's temporary directory (default `/tmp/seatunnel`). The example assumes non-Kerberos HDFS; for Kerberos or HA, configure the additional options from the HdfsFile documentation. The Hadoop CLI used for validation must also be configured to access that cluster. + +## Prepare source data + +:::caution Use an isolated test database + +Run the SQL below once in a MySQL client using a setup account. It intentionally uses `CREATE DATABASE` without `IF NOT EXISTS` and does not drop or truncate any table. If `trade_db` already exists, stop and choose an unused test database name; replace it consistently in the SQL, JDBC URL, `table_path`, and queries. Do not force the SQL client to continue after errors. + +::: + +```sql +CREATE DATABASE trade_db; +USE trade_db; + +CREATE TABLE orders ( + id BIGINT NOT NULL PRIMARY KEY, + order_no VARCHAR(64) NOT NULL, + user_id BIGINT NOT NULL, + amount DECIMAL(10, 2) NOT NULL, + status VARCHAR(32) NOT NULL, + create_time DATETIME NOT NULL +); + +INSERT INTO orders (id, order_no, user_id, amount, status, create_time) VALUES + (1, 'ORD-20260823-001', 10001, 99.50, 'completed', '2026-08-23 10:15:30'), + (2, 'ORD-20260823-002', 10002, 199.00, 'pending', '2026-08-23 14:20:00'), + (3, 'ORD-20260824-001', 10003, 49.90, 'COMPLETED', '2026-08-24 09:00:15'), + (4, 'ORD-20260824-002', 10001, 350.00, 'paid', '2026-08-24 18:45:10'), + (5, 'ORD-20260824-003', 10004, -10.00, 'cancelled', '2026-08-24 20:00:00'); +``` + +The five orders span two dates and include one negative amount to demonstrate filtering. Configure the job account's access to this test table before running SeaTunnel. + +## Complete configuration + +Save the following as `config/mysql-to-hdfs.conf` under your SeaTunnel distribution. + +Replace the JDBC host, database name, `username`, and `password` with your test environment values. Replace `fs.defaultFS` and `path` with your HDFS address and unused test output path. Here, `localhost` means the machine running SeaTunnel, and `namenode` must be resolvable from that machine. The sample credentials are placeholders; use your environment's required TLS settings outside this isolated example. + +```hocon +env { + job.name = "mysql_to_hdfs_batch_dw" + job.mode = "BATCH" + parallelism = 4 +} + +source { + Jdbc { + plugin_output = "src_mysql_orders" + url = "jdbc:mysql://localhost:3306/trade_db?useSSL=false&serverTimezone=UTC&rewriteBatchedStatements=true" + driver = "com.mysql.cj.jdbc.Driver" + username = "test_user" + password = "test_password" + + table_path = "trade_db.orders" + query = "select id, order_no, user_id, amount, status, create_time, date(create_time) as create_date from trade_db.orders" + + partition_column = "id" + partition_num = 4 + partition_lower_bound = 1 + partition_upper_bound = 10000000 + fetch_size = 2000 + } +} + +transform { + Sql { + plugin_input = "src_mysql_orders" + plugin_output = "dwd_orders" + query = """ + select + id as order_id, + order_no, + user_id, + amount, + upper(status) as order_status, + create_time, + FORMATDATETIME(create_date, 'yyyy-MM-dd') as pt_dt + from src_mysql_orders + where amount >= 0 + """ + } +} + +sink { + HdfsFile { + plugin_input = "dwd_orders" + fs.defaultFS = "hdfs://namenode:8020" + path = "/user/hive/warehouse/dwd.db/dwd_orders_df" + file_format_type = "parquet" + partition_by = ["pt_dt"] + compress_codec = "snappy" + schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST" + data_save_mode = "APPEND_DATA" + } +} +``` + +The JDBC query evaluates `date(create_time)` in MySQL. The SQL transform then uses SeaTunnel's `FORMATDATETIME(create_date, 'yyyy-MM-dd')` to build the partition key. Keep `plugin_input` and `plugin_output` consistent between plugins. + +The partition bounds and `partition_num` illustrate JDBC split-read configuration, not performance tuning for five rows. Adapt them to the real source data when scaling up; a tiny dataset need not use all writers or produce equally sized files. + +## Run the job + +Before the first run, record the expected counts in MySQL: + +```sql +SELECT DATE(create_time) AS pt_dt, COUNT(*) AS expected_rows +FROM trade_db.orders +WHERE amount >= 0 +GROUP BY DATE(create_time) +ORDER BY pt_dt; +``` + +For the supplied seed data, expect two rows for each date, four rows in total. Then submit the job: + +```bash +cd "${SEATUNNEL_HOME}" +./bin/seatunnel.sh --config ./config/mysql-to-hdfs.conf -m local +``` + +Wait for the job to finish successfully before checking output. + +:::caution Repeated runs append data + +This job uses `APPEND_DATA`: rerunning it against the same source and output path can add duplicate records. The four-row expectation applies to one successful run into an unused output directory. Use a new test output path for another validation run; do not delete or overwrite existing warehouse data. + +::: + +## Validation result + +### Check the partition directories + +Run the following with your actual HDFS address and output path: + +```bash +hdfs dfs -ls -R hdfs://namenode:8020/user/hive/warehouse/dwd.db/dwd_orders_df +``` + +The expected layout is: + +```text +/user/hive/warehouse/dwd.db/dwd_orders_df/ +├── pt_dt=2026-08-23/ +│ └── <generated-file>.parquet +└── pt_dt=2026-08-24/ + └── <generated-file>.parquet +``` + +The tree is illustrative: the actual names and number of data files depend on writer instances, parallelism, and data distribution. A filename alone does not prove the compression codec. + +### Check the records and file format + +Using a Parquet-capable reader already available in your environment, read all committed data files in both partition directories. For the supplied seed data and a single run, verify these values (row order is not guaranteed): + +| order_id | order_status | amount | pt_dt | +| --- | --- | --- | --- | +| 1 | COMPLETED | 99.50 | 2026-08-23 | +| 2 | PENDING | 199.00 | 2026-08-23 | +| 3 | COMPLETED | 49.90 | 2026-08-24 | +| 4 | PAID | 350.00 | 2026-08-24 | + +Also verify that `order_no`, `user_id`, and `create_time` are preserved. Order `5` must be absent because its amount is negative. Inspect the Parquet metadata to confirm Snappy compression; do not use `cat` to interpret the binary files. + +The `pt_dt` value identifies the directory partition. Depending on the reader, it may need to be inferred from the directory rather than read as a column from an individual file. Hive-style directories do not automatically create or register a Hive table. Directory existence alone is not a complete data validation. + +## Common pitfalls + +- The MySQL JDBC driver is missing from the SeaTunnel process's `lib` directory, or connector versions do not match the distribution. +- `localhost` points to the wrong machine, or the HDFS NameNode/DataNode hostnames are unreachable from SeaTunnel. +- The job account can connect to MySQL but cannot read `trade_db.orders`, or the HDFS identity cannot write output or temporary files. +- The tutorial is run against an existing database or output directory, making the seed data or expected row counts invalid. +- Changing the source query removes `create_date` while the SQL transform still references it. +- A Parquet file is treated as plain text, or a fixed filename/file count is assumed. + +## Related docs + +- [JDBC source](../../connectors/source/Jdbc.md) +- [SQL transform](../../transforms/sql.md) +- [SQL functions](../../transforms/sql-functions.md) +- [HdfsFile sink](../../connectors/sink/HdfsFile.md) diff --git a/docs/en/getting-started/recipes/overview.md b/docs/en/getting-started/recipes/overview.md index 926f72837c..9f4cfc4ed0 100644 --- a/docs/en/getting-started/recipes/overview.md +++ b/docs/en/getting-started/recipes/overview.md @@ -16,6 +16,7 @@ The recipes in this section include concrete prerequisites, complete configurati | CDC from MySQL into Elasticsearch with filtering and field shaping | [MySQL CDC to Elasticsearch](./mysql-cdc-to-elasticsearch.md) | | Batch migration between relational databases with row transformation | [JDBC to JDBC](./jdbc-to-jdbc.md) | | JDBC extraction into object storage | [JDBC to S3](./jdbc-to-s3.md) | +| Batch MySQL extraction into date-partitioned Parquet on HDFS | [MySQL to HDFS](./mysql-to-hdfs.md) | | Streaming from Kafka into Iceberg | [Kafka to Iceberg](./kafka-to-iceberg.md) | | CDC from PostgreSQL into Iceberg | [PostgreSQL CDC to Iceberg](./postgresql-cdc-to-iceberg.md) | | HTTP ingestion into JDBC | [HTTP to JDBC](./http-to-jdbc.md) | diff --git a/docs/sidebars.js b/docs/sidebars.js index 57c1fe4bd9..138386fb9e 100644 --- a/docs/sidebars.js +++ b/docs/sidebars.js @@ -168,6 +168,7 @@ const sidebars = { "getting-started/recipes/mysql-cdc-to-kafka", "getting-started/recipes/mysql-cdc-to-elasticsearch", "getting-started/recipes/jdbc-to-s3", + "getting-started/recipes/mysql-to-hdfs", "getting-started/recipes/kafka-to-iceberg", "getting-started/recipes/postgresql-cdc-to-iceberg", "getting-started/recipes/http-to-jdbc", diff --git a/docs/zh/getting-started/recipes/mysql-to-hdfs.md b/docs/zh/getting-started/recipes/mysql-to-hdfs.md new file mode 100644 index 0000000000..d13858a392 --- /dev/null +++ b/docs/zh/getting-started/recipes/mysql-to-hdfs.md @@ -0,0 +1,215 @@ +--- +title: MySQL 到 HDFS +--- + +# MySQL 到 HDFS + +这条场景教程将 MySQL 订单数据批量写入 HDFS，生成按日期分区、使用 Snappy 压缩的 Parquet 文件。链路由 JDBC Source、SQL Transform 和 HdfsFile Sink 组成。 + +转换步骤将 `id` 重命名为 `order_id`，把订单状态转为大写，过滤负金额，并生成日期分区键 `pt_dt`。这是批量快照示例，不是 CDC 或增量同步任务。 + +## 前置条件 + +1. 先完成 [跑第一个任务](../locally/run-your-first-job.md)。本教程使用 Linux 上的 SeaTunnel Zeta 本地模式。将 `SEATUNNEL_HOME` 设置为 SeaTunnel 发行包的解压目录，不需要检出源码。 +2. 安装与发行包版本一致的 `connector-jdbc` 和 `connector-file-hadoop`。参照 [部署文档](../locally/deployment.md)，在 `config/plugin_config` 中包含以下条目，并保留环境中其他任务需要的连接器。 + +```plugin_config +--seatunnel-connectors-- +connector-jdbc +connector-file-hadoop +--end-- +``` + +3. 将 MySQL JDBC 驱动（例如 `mysql-connector-j-8.x.jar`）放入 `${SEATUNNEL_HOME}/lib`。安装连接器，并检查连接器与驱动是否存在： + +```bash +cd "${SEATUNNEL_HOME}" +sh bin/install-plugin.sh +ls connectors | grep -E 'connector-(jdbc|file-hadoop)' +ls lib | grep 'mysql-connector' +``` + +Zeta 发行包包含 Hadoop jar，应先检查 `lib` 再补充依赖，不要随意混用 Hadoop 客户端版本。环境相关的要求见 [HdfsFile Sink](../../connectors/sink/HdfsFile.md)。 + +4. 准备可访问的 MySQL 实例，以及对源表有 `SELECT` 权限的作业账号。下方初始化 SQL 需要使用具有建库、建表和插入权限的初始化账号；作业账号不需要这些初始化权限。 +5. 准备可访问的 HDFS 集群和未使用的输出目录。SeaTunnel 进程需要对输出目录和 Sink 临时目录（默认 `/tmp/seatunnel`）均有写权限。本例假设 HDFS 未启用 Kerberos；Kerberos 或 HA 环境请按 HdfsFile 文档补充配置。用于验证的 Hadoop CLI 也需要能够访问该集群。 + +## 准备源数据 + +:::caution 使用独立测试数据库 + +在 MySQL 客户端中使用初始化账号执行一次下方 SQL。这里有意使用不带 `IF NOT EXISTS` 的 `CREATE DATABASE`，且不会删除或清空任何表。如果 `trade_db` 已存在，请停止执行，选择未使用的测试库名，并同步替换 SQL、JDBC URL、`table_path` 和查询中的库名。不要强制 SQL 客户端在出错后继续执行。 + +::: + +```sql +CREATE DATABASE trade_db; +USE trade_db; + +CREATE TABLE orders ( + id BIGINT NOT NULL PRIMARY KEY, + order_no VARCHAR(64) NOT NULL, + user_id BIGINT NOT NULL, + amount DECIMAL(10, 2) NOT NULL, + status VARCHAR(32) NOT NULL, + create_time DATETIME NOT NULL +); + +INSERT INTO orders (id, order_no, user_id, amount, status, create_time) VALUES + (1, 'ORD-20260823-001', 10001, 99.50, 'completed', '2026-08-23 10:15:30'), + (2, 'ORD-20260823-002', 10002, 199.00, 'pending', '2026-08-23 14:20:00'), + (3, 'ORD-20260824-001', 10003, 49.90, 'COMPLETED', '2026-08-24 09:00:15'), + (4, 'ORD-20260824-002', 10001, 350.00, 'paid', '2026-08-24 18:45:10'), + (5, 'ORD-20260824-003', 10004, -10.00, 'cancelled', '2026-08-24 20:00:00'); +``` + +这五条订单跨越两个日期，其中一条金额为负数，用于验证过滤效果。运行 SeaTunnel 前，请配置作业账号对该测试表的访问权限。 + +## 完整配置 + +将以下内容保存到 SeaTunnel 发行包内的 `config/mysql-to-hdfs.conf`。 + +将 JDBC 主机、库名、`username` 和 `password` 替换为测试环境的值，将 `fs.defaultFS` 和 `path` 替换为实际 HDFS 地址及未使用的测试输出路径。这里的 `localhost` 指运行 SeaTunnel 的机器，`namenode` 必须能在该机器上解析。示例凭据仅为占位符；在独立测试场景之外，应按环境要求配置 TLS。 + +```hocon +env { + job.name = "mysql_to_hdfs_batch_dw" + job.mode = "BATCH" + parallelism = 4 +} + +source { + Jdbc { + plugin_output = "src_mysql_orders" + url = "jdbc:mysql://localhost:3306/trade_db?useSSL=false&serverTimezone=UTC&rewriteBatchedStatements=true" + driver = "com.mysql.cj.jdbc.Driver" + username = "test_user" + password = "test_password" + + table_path = "trade_db.orders" + query = "select id, order_no, user_id, amount, status, create_time, date(create_time) as create_date from trade_db.orders" + + partition_column = "id" + partition_num = 4 + partition_lower_bound = 1 + partition_upper_bound = 10000000 + fetch_size = 2000 + } +} + +transform { + Sql { + plugin_input = "src_mysql_orders" + plugin_output = "dwd_orders" + query = """ + select + id as order_id, + order_no, + user_id, + amount, + upper(status) as order_status, + create_time, + FORMATDATETIME(create_date, 'yyyy-MM-dd') as pt_dt + from src_mysql_orders + where amount >= 0 + """ + } +} + +sink { + HdfsFile { + plugin_input = "dwd_orders" + fs.defaultFS = "hdfs://namenode:8020" + path = "/user/hive/warehouse/dwd.db/dwd_orders_df" + file_format_type = "parquet" + partition_by = ["pt_dt"] + compress_codec = "snappy" + schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST" + data_save_mode = "APPEND_DATA" + } +} +``` + +JDBC 查询中的 `date(create_time)` 由 MySQL 执行，SQL Transform 再使用 SeaTunnel 的 `FORMATDATETIME(create_date, 'yyyy-MM-dd')` 生成分区键。各插件间的 `plugin_input` 和 `plugin_output` 必须保持一致。 + +分片边界和 `partition_num` 用于展示 JDBC 分片读取配置，不是针对五条记录的性能调优。用于大表时应根据实际源数据调整；少量数据不一定使用全部 Writer，也不保证生成大小相同的文件。 + +## 运行任务 + +第一次运行前，先在 MySQL 中记录预期行数： + +```sql +SELECT DATE(create_time) AS pt_dt, COUNT(*) AS expected_rows +FROM trade_db.orders +WHERE amount >= 0 +GROUP BY DATE(create_time) +ORDER BY pt_dt; +``` + +对于给定样例数据，每个日期应有两条记录，共四条。随后提交任务： + +```bash +cd "${SEATUNNEL_HOME}" +./bin/seatunnel.sh --config ./config/mysql-to-hdfs.conf -m local +``` + +等待任务成功结束后，再检查输出。 + +:::caution 重复运行会追加数据 + +本任务使用 `APPEND_DATA`：对相同源数据和输出路径重复运行，可能追加重复记录。四条记录的预期仅适用于向未使用的输出目录成功运行一次。再次验证时请使用新的测试输出路径，不要删除或覆盖已有数仓数据。 + +::: + +## 验证结果 + +### 检查分区目录 + +使用实际 HDFS 地址和输出路径执行： + +```bash +hdfs dfs -ls -R hdfs://namenode:8020/user/hive/warehouse/dwd.db/dwd_orders_df +``` + +预期目录结构如下： + +```text +/user/hive/warehouse/dwd.db/dwd_orders_df/ +├── pt_dt=2026-08-23/ +│ └── <generated-file>.parquet +└── pt_dt=2026-08-24/ + └── <generated-file>.parquet +``` + +以上目录树仅为示意。实际数据文件的名称和数量取决于 Writer 实例、并行度和数据分布，不能仅凭文件名判断压缩算法。 + +### 检查数据和文件格式 + +使用环境中已有的 Parquet 读取工具，读取两个分区目录内所有已提交的数据文件。对于给定样例数据和单次运行，应核对以下值（不保证行顺序）： + +| order_id | order_status | amount | pt_dt | +| --- | --- | --- | --- | +| 1 | COMPLETED | 99.50 | 2026-08-23 | +| 2 | PENDING | 199.00 | 2026-08-23 | +| 3 | COMPLETED | 49.90 | 2026-08-24 | +| 4 | PAID | 350.00 | 2026-08-24 | + +还应确认 `order_no`、`user_id` 和 `create_time` 保持不变。订单 `5` 因金额为负数应被过滤。通过 Parquet 元数据确认 Snappy 压缩，不要使用 `cat` 将二进制文件当作文本读取。 + +`pt_dt` 表示目录分区值。根据读取工具的行为，可能需要从目录推断该值，而不是从单个文件中读取同名列。Hive 风格目录不会自动创建或注册 Hive 表，仅有目录存在不足以证明数据验证通过。 + +## 常见问题 + +- SeaTunnel 进程使用的 `lib` 目录缺少 MySQL JDBC 驱动，或连接器版本与发行包不一致。 +- `localhost` 指向错误的机器，或 SeaTunnel 无法访问 HDFS NameNode/DataNode 主机。 +- 作业账号可以连接 MySQL，但没有读取 `trade_db.orders` 的权限；或者 HDFS 身份无法写入输出或临时目录。 +- 使用已有数据库或输出目录运行教程，导致样例数据或预期行数不成立。 +- 修改源查询时移除了 `create_date`，但 SQL Transform 仍然引用该字段。 +- 将 Parquet 文件当作纯文本读取，或者假设输出文件名与文件数量固定不变。 + +## 相关文档 + +- [JDBC Source](../../connectors/source/Jdbc.md) +- [SQL Transform](../../transforms/sql.md) +- [SQL 函数](../../transforms/sql-functions.md) +- [HdfsFile Sink](../../connectors/sink/HdfsFile.md) diff --git a/docs/zh/getting-started/recipes/overview.md b/docs/zh/getting-started/recipes/overview.md index 24570dc14f..8d3e85d230 100644 --- a/docs/zh/getting-started/recipes/overview.md +++ b/docs/zh/getting-started/recipes/overview.md @@ -16,6 +16,7 @@ slug: /getting-started/recipes | 从 MySQL CDC 数据实时同步到 Elasticsearch 并完成过滤与字段整形 | [MySQL CDC 到 Elasticsearch](./mysql-cdc-to-elasticsearch.md) | | 从 JDBC 数据批量同步到 JDBC 并进行数据过滤和转换 | [JDBC 到 JDBC](./jdbc-to-jdbc.md) | | 从 JDBC 数据批量同步到 S3 对象存储 | [JDBC 到 S3](./jdbc-to-s3.md) | +| 从 MySQL 批量写入 HDFS，生成按日期分区的 Parquet 文件 | [MySQL 到 HDFS](./mysql-to-hdfs.md) | | 从 Kafka 流式写入 Iceberg | [Kafka 到 Iceberg](./kafka-to-iceberg.md) | | 从 PostgreSQL CDC 数据实时同步到 Iceberg | [PostgreSQL CDC 到 Iceberg](./postgresql-cdc-to-iceberg.md) | | 从 HTTP 数据批量写入 JDBC 关系型数据库 | [HTTP 到 JDBC](./http-to-jdbc.md) |
