SeaTunnel Skill 实战指南用 Claude Code 自然语言驱动数据同步配置生成与智能故障排查【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnelSeaTunnel Skill 是面向 Claude Code 的 AI 集成技能为 Apache SeaTunnel 的操作、配置编写与故障排查提供即时的自然语言交互能力你无需记忆连接器的参数清单只需用一句话描述需求即可获得可运行的 HOCON 配置。本指南以 docs/zh/tools/seatunnel-skill.md 为主体结合当前仓库中 SeaTunnel CLI 内置的技能框架seatunnel-cli/seatunnel_cli/skills/ 与 skills.py展开读完你将掌握 Skill 的安装方法、调用语法、内置技能场景库的结构以及技能 SOP → 黄金示例 → 连接器元数据三层生成框架的底层原理。什么是 SeaTunnel SkillSeaTunnel Skill 是 Claude Code 的 AI 集成技能Skill它把 SeaTunnel 的领域知识以 Markdown 技能文件的形式注入到 Claude Code 的上下文中让 AI 在回答时严格遵循 SeaTunnel 的配置规范与最佳实践。它面向四类典型诉求AI 助手即时获取 SeaTunnel 概念和配置相关帮助知识集成查询官方文档和最佳实践避免凭记忆编造参数智能调试分析任务报错并给出修复建议如 OutOfMemoryError、连接器缺参等代码示例为你的具体用例自动生成配置示例。需要说明的是当前仓库内并没有直接存放Claude Code 版 seatunnel-skill的实现而是把同一套技能思想工程化落地到了 SeaTunnel CLIseatunnel-cli中仓库自带了 8 个结构完全一致的技能场景文件并通过 skills.py 在运行时解析、打分、注入到 LLM 上下文。因此本指南既覆盖官方文档的安装与使用也结合仓库源码把技能文件的编写规范和触发机制讲透。安装与系统要求系统要求已安装 Claude Code本文以官方文档要求为准安装与配置细节请查阅 Claude Code 官方文档Claude Code 技能目录位于~/.claude/skills/。安装步骤官方文档给出的安装方式是从 Apache SeaTunnel 官方工具仓库获取技能文件并复制到 Claude Code 技能目录# 克隆 seatunnel-tools 仓库Apache SeaTunnel 官方工具仓库地址请以官方发布信息为准 git clone seatunnel-tools 仓库地址 cd seatunnel-tools # 复制技能文件到 Claude Code 技能目录 cp -r seatunnel-skill ~/.claude/skills/安装完成后Claude Code 会自动发现~/.claude/skills/seatunnel-skill/目录下的技能并在对话中按需加载。仓库内的替代实践SeaTunnel CLI 技能框架如果你希望在同一套机制下体验技能驱动配置生成当前仓库的 SeaTunnel CLI 已经内置了完整的技能框架无需手动复制文件。从源码方式安装cd seatunnel-cli bash setup.sh # 安装所有 LLM 提供商 开发工具 seatunnel --init # 配置 LLM 提供商或直接使用 SeaTunnel 发行版自带的封装脚本首次运行自动安装依赖bin/seatunnel-ai.sh --init bin/seatunnel-ai.shCLI 的技能文件位于 seatunnel-cli/seatunnel_cli/skills/包含 8 个场景技能batch_sync、cdc_realtime、conditional_routing、cross_database、data_quality、file_etl、multi_pipeline、transform_chain。二者理念一致用结构化的技能 SOP 约束 AI 的生成行为而不是放任 LLM 自由发挥。使用方法在 Claude Code 中调用技能安装完成后直接在 Claude Code 中以斜杠命令形式调用# 查询 SeaTunnel 文档 /seatunnel-skill 如何配置 MySQL 到 PostgreSQL 的数据同步 # 获取连接器信息 /seatunnel-skill 列出所有可用的 Kafka 连接器选项 # 调试配置问题 /seatunnel-skill 为什么我的任务出现 OutOfMemoryError 错误 # 生成配置示例 /seatunnel-skill 创建一个 MySQL 到 Elasticsearch 的任务配置四个示例恰好对应技能的四种核心能力文档查询知识集成、连接器信息获取知识集成、错误诊断智能调试、配置生成代码示例。其中列出 Kafka 连接器选项和生成配置两类请求在 CLI 中还有更精细的执行链路——CLI 会通过list_connectors/get_connector_info两个工具定义于 agents.py先从连接器知识库中检索真实参数再让生成 Agent 只使用检索到的参数名从而杜绝编造参数名。在 CLI 中等价的能力通过单次命令即可完成# 生成并展示配置 seatunnel Sync MySQL users table to S3 Parquet # 生成并保存到文件 seatunnel 从 Kafka 读取订单数据写入 ClickHouse -o my_job.conf # 动态切换 LLM 提供商 seatunnel Read CSV files and write to Elasticsearch --provider openai --model gpt-4o技能场景库8 个内置 Skill 的职责与配置模式仓库中的每个技能文件都遵循统一的 YAML frontmatter Markdown 正文结构。以 batch_sync.md 为例frontmatter 声明了技能的名称、描述、触发词和可用工具--- name: batch_sync description: Batch data synchronization from source to sink (one-time or scheduled full/incremental load) triggers: - batch - sync - migrate - import - export - dump - load - copy - transfer - extract tools: - get_connector_info - list_connectors composable: true ---其中triggers是技能触发匹配的关键运行时会把用户请求文本与每个技能的触发词做包含匹配并打分见 skills.py得分最高的技能最多注入 2 个受_MAX_SKILLS限制连同其 Domain Knowledge / SOP / Constraints / Pattern 一起注入 LLM 上下文。若没有匹配到模式类技能cdc_realtime、file_etl、cross_database、data_quality则默认回退到batch_sync。batch_sync批量数据同步默认技能适用于全表同步、基于 SELECT 的抽取、一次性迁移、定时 ETL 加载当用户请求中不包含流式/CDC/实时关键词时它就是默认技能。核心领域知识批处理模式在env块中设置job.mode BATCHparallelism控制并发读写线程数默认 2 较为安全大表可调高JDBC 源用query做灵活抽取用table_path做自动分片并行读取多表批量同步时每张表需要独立的 source 块和唯一的plugin_output路由标签凭据必须使用${ENV_VAR}占位符绝不硬编码密码source 的plugin_output与 sink 的plugin_input负责把数据路径串联起来。标准配置模式env { parallelism parallelism job.mode BATCH } source { SourceConnector { source_options plugin_output routing_label } } sink { SinkConnector { sink_options plugin_input routing_label } }cdc_realtimeCDC 实时变更数据同步适用于使用 CDC 连接器MySQL-CDC、PostgreSQL-CDC、MongoDB-CDC、Oracle-CDC、SqlServer-CDC 等的持续实时同步数据以 insert/update/delete 事件流的形式源源不断流动。技能文件 cdc_realtime.md 特别强调了几条与批处理截然不同的约束CDC 模式必须使用job.mode STREAMING绝不使用 BATCHcheckpoint.interval毫秒控制状态检查点频率默认 1000010 秒对大多数场景安全CDC 源的参数键使用连字符database-name/table-name不是下划线table-name支持正则如mydb\\.orders或mydb\\..*CDC 源不使用query它捕获的是整表变更日志sink 必须支持 upsert/delete 语义如 StarRocks、Doris、带主键的 JdbcCDC 才能正确工作MySQL-CDC 需要数据库权限 REPLICATION SLAVE、REPLICATION CLIENT。技能文件还整理了 MySQL-CDC 与 PostgreSQL-CDC 的差异对照选项MySQL-CDCPostgreSQL-CDC流来源binlog逻辑复制流WALslot.name不适用建议显式设置每个并发作业一个 slotdecoding.plugin.name不适用pgoutputPG 10 内置table-names条目格式database.tableschema.table规范形式服务端前置条件binlog ROW 模式wal_level logical标准配置模式env { parallelism parallelism job.mode STREAMING checkpoint.interval 10000 } source { CDC-Connector { hostname host port port username ${DB_USER} password ${DB_PASSWORD} database-names [database] # MySQL-CDC 条目: database.table; PostgreSQL-CDC: schema.table table-names [database-or-schema-qualified-table] plugin_output routing_label } } sink { SinkConnector { sink_options plugin_input routing_label } }transform_chain转换链适用于在 source 与 sink 之间做数据转换如行过滤、字段变换、SQL 表达式、字段映射、数据脱敏、LLM 增强等。技能文件 transform_chain.md 把转换插件归为四类SQL 类Sql最灵活支持 WHERE、类 JOIN 操作、聚合、类型转换能用 SQL 表达的需求优先用它字段操作类FieldMapper重命名/重排/删除字段、Copy复制字段、Split按分隔符拆分、Replace正则替换、Filter白名单选字段、FilterRowKind按 INSERT / UPDATE_BEFORE / UPDATE_AFTER / DELETE 变更类型过滤CDC 源后常用AI 类LLM调用 LLM API 做分类/抽取/摘要等数据增强、Embedding为文本字段生成向量数据质量类JsonPath从 JSON 字符串字段提取值、DynamicCompile自定义 Java 代码转换高级用法。转换链通过plugin_input/plugin_output逐跳串联source 输出到标签 A → Transform 1 从 A 读、输出到 B → Transform 2 从 B 读、输出到 C → sink 从 C 读。每个 transform 块必须同时具有plugin_input和plugin_outputenv { parallelism parallelism job.mode mode } source { SourceConnector { source_options plugin_output src_label } } transform { TransformPlugin { plugin_input src_label plugin_output transform_label transform_options } } sink { SinkConnector { sink_options plugin_input transform_label } }multi_pipeline单作业多管道适用于在一个 SeaTunnel 作业中承载多条相互独立的数据路径例如表 A 同步到 Console、表 B 同步到 Assert或从 2 个源读取写入不同 sink。技能文件 multi_pipeline.md 强调的规则包括所有 source 块必须放在同一个source { }段内禁止创建多个source { }顶层段sink 同理每个plugin_output值在所有 source 块中必须唯一每个plugin_input必须恰好匹配一个plugin_output同一连接器类型可以出现多个块如两个 Jdbc source 读不同表管道展开Pipeline Expansion当 sink 不支持多表输入如 Console、Assert、Clickhouse时1 个管道同步 3 张表需要展开为 3 个 source 块 3 个 sink 块而 Jdbcgenerate_sink_sql true或原生支持多表语法的 CDC正则table-name则无需展开。conditional_routing条件分流适用于一个源按行级条件分发到不同 sink例如金额 1000 写文件其余写 console、ERROR 日志进 console其余归档、已支付订单进 A、未支付进 B。技能文件 conditional_routing.md 指出这不是多管道独立源也不是扇出同样的行到所有 sink而是单一数据流按互斥谓词拆分拆分由多个并行 Sql transform 消费同一个 source 输出实现每个 transform 声明plugin_input source_label并携带各自的 WHERE 谓词输出到不同的plugin_outputSeaTunnel 会自动把源流复制给所有引用其标签的 transform源上无需额外配置谓词必须互斥且尽量完备amount 100应配对amount 100而不是amount 100否则等值行会被静默丢弃受 Zeta SQL transform 能力限制仅支持投影 WHERE不支持 GROUP BY、JOIN、ORDER BY。cross_database跨数据库同步适用于不同数据库系统之间的数据同步MySQL → PostgreSQL、Oracle → MySQL、SQL Server → PostgreSQL 等核心挑战是处理不同的 JDBC 驱动、URL 格式、凭据集合与类型映射差异。技能文件 cross_database.md 给出的关键知识源与 sink 必须使用不同的${ENV_VAR}凭据如${SOURCE_DB_USER}与${SINK_DB_USER}它们是不同的系统常见 JDBC URL 模板与驱动类MySQLjdbc:mysql://host:port/database/com.mysql.cj.jdbc.DriverPostgreSQLjdbc:postgresql://host:port/database/org.postgresql.DriverOraclejdbc:oracle:thin:host:port:sid/oracle.jdbc.driver.OracleDriverSQL Serverjdbc:sqlserver://host:port;databaseNamedatabase/com.microsoft.sqlserver.jdbc.SQLServerDriverJdbc sink 上设置generate_sink_sql true可启用自动 DDL目标表不存在时 SeaTunnel 会自动建表类型映射由 SeaTunnel 内部类型系统自动处理但 Oracle NUMBER、SQL Server NVARCHAR 等边界情况需要关注。data_quality数据质量校验适用于校验数据质量——检查行数、字段值、数据类型或用 Assert sink 验证管道输出是否符合预期也适用于基于 FakeSource 的测试场景。技能文件 data_quality.md 的核心知识Assertsink 按规则校验数据规则不满足则作业失败rules { }块中row_rules校验行数MIN_ROW、MAX_ROWfield_rules校验字段值NOT_NULL、MIN、MAX、MIN_LENGTH、MAX_LENGTHFakeSource生成带可配置 schema 的合成测试数据tables_configs数组包含row.num与schema { fields { ... } }注意是带下划线的tables_configs不是table_configConsolesink 用于调试/检查把数据打印到标准输出生产环境的数据质量门禁推荐真实 source Assert sink组合在装载到实际 sink 之前先做校验需要边校验边装载时用 multi_pipeline 技能把数据同时路由到 Assert 与真实 sink。env { parallelism 1 job.mode BATCH } source { FakeSource { tables_configs [ { row.num row_count schema { fields { field_name type } } } ] plugin_output routing_label } } sink { Assert { plugin_input routing_label rules { row_rules [ { rule_type MIN_ROW rule_value expected_min } ] } } }技能框架的底层原理三层生成与多 Agent 流水线SeaTunnel CLI 把技能放进了完整的多 Agent 生成流水线中这也解释了为什么技能文件的效果远胜于直接把文档塞给 LLM。三层生成框架Skill SOP → Golden Example → Connector MetadataCLI READMEseatunnel-cli/README.md明确描述了三层生成架构Skill SOP第一层从 skills/ 加载匹配的技能注入其 Domain Knowledge、SOP、Constraints 与 Pattern约束 AI 遵循既定套路Golden Example第二层若存在 (source, sink) 组合的黄金示例位于 golden_examples/以其作为结构参照Connector Metadata第三层通过get_connector_info工具检索真实连接器参数规则只使用元数据中存在的参数名。第三层的连接器知识库采用两层解析 智能回退优先使用运行中 SeaTunnel 引擎的/option-rules接口返回的实时元数据引擎不可用时回退到随包分发的connector_metadata.json由 SeaTunnel 引擎通过反射导出覆盖 150 连接器的选项规则与值约束零 LLM token 成本。多 Agent 流水线从 agents.py 与 README 的架构图可以还原完整的调用链用户自然语言输入 ↓ Planner Agent分析意图 查询连接器知识库→ 生成结构化计划 ↓ Config Agent依据 Skill SOP / Golden Example / Connector Metadata 生成 HOCON ↓ Validator Agent本地校验语法、结构、必填参数、括号匹配、安全检查 ↓ 引擎 seatunnel.sh --check 可选 REST API 校验 通过 ── 是 ──→ 输出 自动保存到 .data/last_job.conf ↓ 否最多 3 轮 Fix Agent自动纠错 ↓ /run 或 /check 失败时 Repair Agent诊断 修复配置本地校验会标记未解析的${VAR}占位符为缺失环境变量但对 SeaTunnel 引擎在文件 sink 模板中解析的占位符做了字段级豁免file_name_expression允许${now}、${uuid}、${transactionId}partition_dir_expression允许${k0}、${v0}等分区占位符同一名称用在其他字段URL、凭据、路径时仍按普通环境变量处理并报告未设置。最佳实践与注意事项结合技能文件中的 Constraints 与 CLI 的工程实现以下实践可以显著提升配置质量凭据一律使用环境变量占位符所有技能文件都强调Credentials always as ${ENV_VAR}跨库同步时源与 sink 还要分开命名${SOURCE_DB_USER}/${SINK_DB_USER}CLI 的本地校验和/remember记忆系统都会拦截明文密码与 API Key路由标签是配置的接线图先列出所有 (plugin_output → plugin_input) 配对再生成配置避免出现孤立标签、标签复用两个 source 输出同一标签和transform 被静默旁路sink 仍消费旧标签模式选择要果断批量同步用job.mode BATCH且不加checkpoint.intervalCDC 用job.mode STREAMING且必须带checkpoint.interval二者不可混用参数名以元数据为准不要凭记忆写参数尤其注意 CDC 源的连字符键database-name与下划线键FakeSource 的tables_configs这类容易混淆的差异一律以get_connector_info返回为准善用校验闭环生成后依次走本地校验 → 引擎--check→ REST API 校验失败时交给自动修复 Agent而不是手工逐行排查。小结SeaTunnel Skill 把数据同步领域的专家经验封装成了可复用的技能文件对 Claude Code 用户而言它是开箱即用的/seatunnel-skill命令对开发者而言当前仓库 seatunnel-cli/seatunnel_cli/skills/ 下的 8 个技能文件展示了同一套思想的完整工程化实现——YAML frontmatter 声明触发词与工具、Markdown 正文承载领域知识与 SOP、技能匹配引擎负责按需注入。理解这套机制后你既可以熟练使用现成的技能完成 MySQL 到 PostgreSQL 同步、CDC 实时同步、条件分流、跨库迁移等常见任务也可以参考现有技能文件的写法为自己的场景定制新的技能。【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
