Apache SeaTunnel脚本升级及参数调优实战

最近作者针对实时数仓的Apache SeaTunnel同步链路，完成了双引擎架构升级与全链路参数深度调优，希望本文能够给大家有所启发，欢迎品评指正！

   Apache SeaTunnel 版本 ：2.3.9
  Doris版本：2.0.6
  MySQL JDBC Connector : 8.0.28
  架构升级

• 批处置处罚链路：JDBC并行度进行提拔，基于ID分区实现分片读取，联合批量参数（fetch_size=10000+batch_size=5000）使全量同步吞吐量大幅增加
  
• 实时增量链路：引入MySQL-CDC组件，通过initial快照模式+chunk.size.rows=8096实现全量/增量平滑切换，变乱延长压降至500ms内
  

稳定性加强

• 资源管控：JDBC连接池动态扩容（max_size=20）+ CDC限流计谋（rows_per_second=1000），源库CPU峰值负载下降40%
  
• 容错机制：Doris两阶段提交（enable-2pc=true）共同检查点（checkpoint.interval=10s），故障规复时间收缩80%
  

写入优化

• 缓冲区三级联控（buffer-size=10000+buffer-count=3+flush.interval=5s）提拔Doris写入批次质量
  
• Tablet粒度控制（request_tablet_size=5）使BE节点负载均衡度提拔
  

实战演示

同步之前创建Doris表

1. -- DROP TABLE IF EXISTS ods.ods_activity_info_full;
2. CREATE TABLE ods.ods_activity_info_full
3. (
4.     `id`            VARCHAR(255) COMMENT '活动id',
5.     `k1`            DATE NOT NULL   COMMENT '分区字段',
6.     `activity_name` STRING COMMENT '活动名称',
7.     `activity_type` STRING COMMENT '活动类型',
8.     `activity_desc` STRING COMMENT '活动描述',
9.     `start_time`    STRING COMMENT '开始时间',
10.     `end_time`      STRING COMMENT '结束时间',
11.     `create_time`   STRING COMMENT '创建时间'
12. )
13.     ENGINE=OLAP  -- 使用Doris的OLAP引擎，适用于高并发分析场景
14.     UNIQUE KEY(`id`,`k1`)  -- 唯一键约束，保证(id, k1)组合的唯一性（Doris聚合模型特性）
15. COMMENT '活动信息全量表'
16. PARTITION BY RANGE(`k1`) ()  -- 按日期范围分区（具体分区规则由动态分区配置决定）
17. DISTRIBUTED BY HASH(`id`)  -- 按id哈希分桶，保证相同id的数据分布在同一节点
18. PROPERTIES
19. (
20.     "replication_allocation" = "tag.location.default: 1",  -- 副本分配策略：默认标签分配1个副本
21.     "is_being_synced" = "false",          -- 是否处于同步状态（通常保持false）
22.     "storage_format" = "V2",             -- 存储格式版本（V2支持更高效压缩和索引）
23.     "light_schema_change" = "true",      -- 启用轻量级schema变更（仅修改元数据，无需数据重写）
24.     "disable_auto_compaction" = "false", -- 启用自动压缩（合并小文件提升查询性能）
25.     "enable_single_replica_compaction" = "false", -- 禁用单副本压缩（多副本时保持数据一致性）
27.     "dynamic_partition.enable" = "true",            -- 启用动态分区
28.     "dynamic_partition.time_unit" = "DAY",          -- 按天创建分区
29.     "dynamic_partition.start" = "-60",             -- 保留最近60天的历史分区
30.     "dynamic_partition.end" = "3",                 -- 预先创建未来3天的分区
31.     "dynamic_partition.prefix" = "p",              -- 分区名前缀（如p20240101）
32.     "dynamic_partition.buckets" = "32",            -- 每个分区的分桶数（影响并行度）
33.     "dynamic_partition.create_history_partition" = "true", -- 自动创建缺失的历史分区
35.     "bloom_filter_columns" = "id,activity_name",  -- 为高频过滤字段(id/名称)创建布隆过滤器，加速WHERE查询
36.     "compaction_policy" = "time_series",          -- 按时间序合并策略优化时序数据（适合活动时间字段）
37.     "enable_unique_key_merge_on_write" = "true",  -- 唯一键写时合并（实时更新场景减少读放大）
38.     "in_memory" = "false"                        -- 关闭全内存存储（仅小表可开启）
39. );

复制代码

设置SeaTunnel JDBC同步脚本

1. env {
2.   # 环境配置
3.   parallelism = 8                     # 增加并行度以提高吞吐量
4.   job.mode = "STREAMING"              # 使用流式处理模式进行实时同步
5.   checkpoint.interval = 10000         # 检查点间隔，单位毫秒
7.   # 限流配置 - 避免对源数据库造成过大压力
8.   read_limit.bytes_per_second = 10000000  # 每秒读取字节数限制，约10MB/s
9.   read_limit.rows_per_second = 1000       # 每秒读取行数限制
11.   # 本地检查点配置
12.   execution.checkpoint.data-uri = "file:///opt/seatunnel/checkpoints"
13.   execution.checkpoint.max-concurrent = 1  # 最大并发检查点数
15.   # 性能优化参数
16.   execution.buffer-timeout = 5000          # 缓冲超时时间(毫秒)
17.   execution.jvm-options = "-Xms4g -Xmx8g -XX:+UseG1GC -XX:MaxGCPauseMillis=100"
18. }
20. source {
21.   MySQL-CDC {
22.     # 基本连接配置
23.     # server-id = 5652-5657             # MySQL复制客户端的唯一ID范围
24.     username = "root"                # 数据库用户名
25.     password = ""                # 数据库密码
26.     table-names = ["gmall.activity_info"]  # 要同步的表
27.     base-url = "jdbc:mysql://192.168.241.128:3306/gmall?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai"
29.     # CDC 特有配置
30.     schema-changes.enabled = true     # 启用架构变更捕获
31.     server-time-zone = "Asia/Shanghai"  # 服务器时区
33.     # 性能优化配置
34.     snapshot.mode = "initial"         # 初始快照模式
35.     snapshot.fetch.size = 10000       # 快照获取大小
36.     chunk.size.rows = 8096            # 分块大小，用于并行快照
37.     connection.pool.size = 10         # 连接池大小
39.     # 高级配置
40.     include.schema.changes = true     # 包含架构变更事件
41.     scan.startup.mode = "initial"     # 启动模式：initial(全量+增量)
42.     scan.incremental.snapshot.chunk.size = 8096  # 增量快照分块大小
43.     debezium.min.row.count.to.stream.results = 1000  # 流式结果的最小行数
45.     # 容错配置
46.     connect.timeout = 30000           # 连接超时时间(毫秒)
47.     connect.max-retries = 3           # 最大重试次数
49.     # 输出表名
50.     result_table_name = "mysql_cdc_source"
51.   }
52. }
54. # 可选的转换逻辑，如果需要对数据进行处理
55. transform {
56.   Sql {
57.     source_table_name = "mysql_cdc_source"
58.     result_table_name = "doris_sink_data"
60.     # 根据需要转换字段，这里添加了一个分区字段k1
61.     query = """
62.       select
63.         id,
64.         formatdatetime(create_time,'yyyy-MM-dd') as k1,
65.         activity_name,
66.         activity_type,
67.         activity_desc,
68.         start_time,
69.         end_time,
70.         create_time
71.       from mysql_cdc_source
72.     """
73.   }
74. }
76. sink {
77.   Doris {
78.     # 基本连接配置
79.     source_table_name = "doris_sink_data"  # 或直接使用 "mysql_cdc_source"
80.     fenodes = "192.168.241.128:8030"
81.     username = "root"
82.     password = ""
83.     table.identifier = "ods.ods_activity_info_full"  # Doris目标表
85.     # 事务和标签配置
86.     sink.enable-2pc = "true"          # 启用两阶段提交，确保一致性
87.     sink.label-prefix = "cdc_sync"    # 导入标签前缀
89.     # 写入模式配置
90.     sink.properties {
91.       format = "json"
92.       read_json_by_line = "true"
93.       column_separator = "\t"         # 列分隔符
94.       line_delimiter = "\n"           # 行分隔符
95.       max_filter_ratio = "0.1"        # 允许的最大错误率
97.       # CDC特有配置 - 处理不同操作类型
98.       # 使用Doris的UPSERT模式处理CDC事件
99.       merge_type = "MERGE"            # 合并类型：APPEND或MERGE
100.       delete_enable = "true"          # 启用删除操作
101.     }
103.     # 性能优化配置
104.     sink.buffer-size = 10000          # 缓冲区大小
105.     sink.buffer-count = 3             # 缓冲区数量
106.     sink.flush.interval-ms = 5000     # 刷新间隔
107.     sink.max-retries = 3              # 最大重试次数
108.     sink.parallelism = 8              # 写入并行度
110.     # Doris连接优化
111.     doris.config = {
112.       format = "json"
113.       read_json_by_line = "true"
114.       request_connect_timeout_ms = "5000"  # 连接超时
115.       request_timeout_ms = "30000"         # 请求超时
116.       request_tablet_size = "5"            # 每个请求的tablet数量
117.     }
118.   }
119. }

复制代码

设置SeaTunnel MySQLCDC 同步脚本

1. env {
2.   parallelism = 8
3.   job.mode = "BATCH"
4.   checkpoint.interval = 30000
6.   # 本地文件系统检查点
7.   execution.checkpoint.data-uri = "file:///opt/seatunnel/checkpoints"
8.   execution.buffer-timeout = 5000
10.   # JVM 参数优化
11.   execution.jvm-options = "-Xms4g -Xmx8g -XX:+UseG1GC -XX:MaxGCPauseMillis=100"
12. }
14. source {
15.   Jdbc {
16.     result_table_name = "mysql_seatunnel"
17.     url = "jdbc:mysql://192.168.241.128:3306/gmall?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useSSL=false&rewriteBatchedStatements=true&useServerPrepStmts=true&cachePrepStmts=true"
18.     driver = "com.mysql.cj.jdbc.Driver"
19.     connection_check_timeout_sec = 30
20.     user = "gmall"
21.     password = "gmall"
23.     # 使用分区并行读取
24.     query = "select id, activity_name, activity_type, activity_desc, start_time, end_time, create_time from gmall.activity_info"
25.     partition_column = "id"
26.     partition_num = 8
28.     # 连接池配置
29.     connection_pool {
30.       max_size = 20
31.       min_idle = 5
32.       max_idle_ms = 60000
33.     }
35.     # 批处理配置
36.     fetch_size = 10000
37.     batch_size = 5000
38.     is_exactly_once = true
39.   }
40. }
42. transform {
43.   Sql {
44.     source_table_name = "mysql_seatunnel"
45.     result_table_name = "seatunnel_doris"
47.     query = """
48.       select
49.         id,
50.         formatdatetime(create_time,'yyyy-MM-dd') as k1,  
51.         activity_name,
52.         activity_type,
53.         activity_desc,
54.         start_time,
55.         end_time,
56.         create_time
57.       from mysql_seatunnel
58.     """
59.   }
60. }
62. sink {
63.   Doris {
64.     source_table_name = "seatunnel_doris"
65.     fenodes = "192.168.241.128:8030"
66.     username = "root"
67.     password = ""
68.     table.identifier = "ods.ods_activity_info_full"
69.     sink.enable-2pc = "true"
70.     sink.label-prefix = "test_json"
72.     # 优化Doris写入配置
73.     sink.properties {
74.       format = "json"
75.       read_json_by_line = "true"
76.       column_separator = "\t"
77.       line_delimiter = "\n"
78.       max_filter_ratio = "0.1"
79.     }
81.     # 批量写入配置
82.     sink.buffer-size = 10000
83.     sink.buffer-count = 3
84.     sink.flush.interval-ms = 5000
85.     sink.max-retries = 3
86.     sink.parallelism = 8
88.     doris.config = {
89.       format = "json"
90.       read_json_by_line = "true"
91.       request_connect_timeout_ms = "5000"
92.       request_timeout_ms = "30000"
93.       request_tablet_size = "5"
94.     }
95.   }
96. }

复制代码

终极Apache Doris数据：

本文完！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。