--- name: data-sync-sql description: 从业务数据库同步数据到数据仓库，生成全量、增量同步SQL语句 license: MIT compatibility: opencode metadata: audience: data engineers, ETL developers workflow: data synchronization target_scenario: business DB to data warehouse sync --- ## 功能概述 本技能根据业务数据库表结构，生成将数据同步到数据仓库的SQL语句。技能严格遵循以下原则： 1. 生成全量同步SQL语句 2. 生成增量同步SQL语句 3. 生成全量和增量统计记录数的SQL 4. 使用固定的增量参数名称：`sync_last_time` 5. 增量字段统一为日期字段，通过查询业务数据确定具体字段 6. 控制登账字段必须大于1的记录才能同步 7. 输出JSON格式的同步定义，包含表信息、字段信息、SQL代码 8. 生成的JSON文件保存到项目`sync-sql`目录下 9. 所有SQL都包含分页参数 10. 支持调用工具测试SQL语法 ## 使用前准备 在使用本技能前，请确保以下信息已明确： 1. **业务表来源**：业务表定义的位置（Go结构体文件、数据库连接、SQL文件等） 2. **数据仓库表信息**：目标数据仓库表名和结构 3. **同步需求**：全量同步、增量同步或两者都需要 4. **项目路径**：生成的SQL文件保存的项目目录（默认为当前项目的`sync-sql`目录） ## 工作流程 ### 1. 确定业务表和目标表 - 询问用户业务表名称和目标数据仓库表名 - 如果涉及多表关联，询问关联关系和JOIN条件 - 使用`glob`和`read`工具读取业务表定义文件 - 解析业务表字段结构，包括字段名、数据类型、注释等 ### 2. 识别增量字段 - 查询业务表结构，找出所有日期/时间类型的字段 - 使用数据分析工具或查询示例数据，确定最适合作为增量同步的字段 - 优先选择以下字段作为增量字段： - `updated_at`（更新时间） - `created_at`（创建时间） - `modified_date`（修改日期） - 其他业务日期字段（如`order_date`、`transaction_date`） - 如果无法自动确定，使用提问工具向用户确认增量字段 ### 3. 识别登账字段 - 查询业务表注释和字段定义，找出表示"登账状态"或"有效记录"的字段 - 常见登账字段名称：`status`、`is_valid`、`is_active`、`approval_status`等 - 通过查询示例数据或询问用户，确定登账字段和有效值条件 - 默认规则：登账字段值必须大于1的记录才能同步（如`status > 1`） ### 4. 生成全量同步SQL - 基础SELECT语句选择所有字段 - 添加WHERE条件：登账字段 > 1 - 添加分页参数：`LIMIT {page_size} OFFSET {offset}` - 生成完整的INSERT INTO或MERGE语句（根据数据仓库类型） - 包含数据仓库表名和字段映射 ### 5. 生成增量同步SQL - 使用固定的增量参数名称：`sync_last_time` - 基础SELECT语句选择所有字段 - 添加WHERE条件： - 登账字段 > 1 - 增量字段 > :sync_last_time - 添加分页参数 - 按增量字段升序排序，确保数据顺序 - 生成完整的INSERT或MERGE语句 ### 6. 生成统计SQL - 全量统计：`SELECT COUNT(*) FROM 业务表 WHERE 登账字段 > 1` - 增量统计：`SELECT COUNT(*) FROM 业务表 WHERE 登账字段 > 1 AND 增量字段 > :sync_last_time` - 统计SQL用于监控同步进度和计算分页 ### 7. 生成JSON输出 - 按照以下JSON格式组织输出： ```json { "warehouse_table": "数据仓库表名称", "business_tables": "业务数据表名称，多个用逗号隔开", "description": "同步任务描述", "fields": [ { "name": "字段名", "source_name": "源字段名", "type": "字段类型", "description": "字段中文描述", "is_incremental": "是否增量字段", "is_accounting": "是否登账字段" } ], "full_sync_sql": "全量同步SQL代码", "incremental_sync_sql": "增量同步SQL代码", "count_sql": { "full_count": "全量统计SQL", "incremental_count": "增量统计SQL" }, "parameters": { "incremental_param": "sync_last_time", "page_size_param": "page_size", "offset_param": "offset" }, "config": { "accounting_field": "登账字段名", "accounting_condition": "登账条件", "incremental_field": "增量字段名" } } ``` ### 8. 保存文件 - 在项目根目录下创建`sync-sql`目录（如果不存在） - 生成文件名格式：`{业务表名}_sync_{数据仓库表名}.json` - 使用`write`工具将JSON内容写入文件 - 同时生成一个纯SQL文件（可选）：`{业务表名}_sync_{数据仓库表名}.sql` ### 9. 测试SQL语法 - 调用Bash工具测试生成的SQL语法 - 如果有数据库测试环境，使用相应客户端进行验证 - 如果没有，使用`mysql --parse`进行基本语法检查 - 报告任何语法错误并修正 ## SQL模板示例 ### 全量同步SQL模板 ```sql -- 全量同步：{业务表} -> {数据仓库表} -- 分页参数：:page_size, :offset INSERT INTO {数据仓库表} ({字段列表}) SELECT {字段映射列表} FROM {业务表} WHERE {登账字段} > 1 ORDER BY {主键字段} LIMIT :page_size OFFSET :offset; ``` ### 增量同步SQL模板 ```sql -- 增量同步：{业务表} -> {数据仓库表} -- 参数：:sync_last_time, :page_size, :offset INSERT INTO {数据仓库表} ({字段列表}) SELECT {字段映射列表} FROM {业务表} WHERE {登账字段} > 1 AND {增量字段} > :sync_last_time ORDER BY {增量字段} ASC, {主键字段} LIMIT :page_size OFFSET :offset; ``` ### 统计SQL模板 ```sql -- 全量统计 SELECT COUNT(*) as total_count FROM {业务表} WHERE {登账字段} > 1; -- 增量统计 SELECT COUNT(*) as incremental_count FROM {业务表} WHERE {登账字段} > 1 AND {增量字段} > :sync_last_time; ``` ## 多表关联支持 如果同步需要多表关联，技能支持以下模式： ### 多表JOIN示例 ```sql -- 多表关联同步 INSERT INTO dw_order_details ({字段列表}) SELECT o.order_id, o.order_date, c.customer_name, p.product_name, o.quantity, o.amount FROM orders o JOIN customers c ON o.customer_id = c.customer_id JOIN products p ON o.product_id = p.product_id WHERE o.status > 1 AND o.updated_at > :sync_last_time ORDER BY o.updated_at ASC LIMIT :page_size OFFSET :offset; ``` ## 参数说明 ### 固定参数名称 1. **增量参数**：`sync_last_time` - 上次同步时间点 2. **分页大小**：`page_size` - 每页记录数（建议值：1000） 3. **偏移量**：`offset` - 分页偏移量 ### 配置参数 1. **登账字段**：自动识别或用户指定的状态字段 2. **登账条件**：固定为 `> 1`（可根据需求调整） 3. **增量字段**：自动识别的最佳日期字段 ## 输出示例 ### JSON格式输出 ```json { "warehouse_table": "dw_user_info", "business_tables": "user_info, user_profile", "description": "用户信息同步到数据仓库", "fields": [ { "name": "user_id", "source_name": "id", "type": "BIGINT", "description": "用户ID", "is_incremental": false, "is_accounting": false }, { "name": "user_name", "source_name": "name", "type": "VARCHAR(100)", "description": "用户姓名", "is_incremental": false, "is_accounting": false }, { "name": "status", "source_name": "status", "type": "TINYINT", "description": "用户状态", "is_incremental": false, "is_accounting": true }, { "name": "updated_at", "source_name": "update_time", "type": "DATETIME", "description": "更新时间", "is_incremental": true, "is_accounting": false } ], "full_sync_sql": "INSERT INTO dw_user_info (user_id, user_name, status, updated_at)\nSELECT \n id as user_id,\n name as user_name,\n status,\n update_time as updated_at\nFROM user_info\nWHERE status > 1\nORDER BY id\nLIMIT :page_size OFFSET :offset;", "incremental_sync_sql": "INSERT INTO dw_user_info (user_id, user_name, status, updated_at)\nSELECT \n id as user_id,\n name as user_name,\n status,\n update_time as updated_at\nFROM user_info\nWHERE status > 1\n AND update_time > :sync_last_time\nORDER BY update_time ASC, id\nLIMIT :page_size OFFSET :offset;", "count_sql": { "full_count": "SELECT COUNT(*) as total_count FROM user_info WHERE status > 1;", "incremental_count": "SELECT COUNT(*) as incremental_count FROM user_info WHERE status > 1 AND update_time > :sync_last_time;" }, "parameters": { "incremental_param": "sync_last_time", "page_size_param": "page_size", "offset_param": "offset" }, "config": { "accounting_field": "status", "accounting_condition": "> 1", "incremental_field": "update_time" } } ``` ## 注意事项 1. **增量参数固定**：始终使用 `sync_last_time` 作为增量参数名称 2. **登账字段控制**：所有同步都必须包含 `{登账字段} > 1` 条件 3. **日期字段识别**：自动识别最适合的日期字段作为增量字段 4. **分页处理**：所有SQL都包含分页参数，避免大数据量内存问题 5. **字段映射**：正确处理业务表和数据仓库表的字段名差异 6. **数据类型兼容**：确保字段类型在源和目标之间兼容 7. **性能优化**：增量SQL按增量字段排序，便于顺序读取 ## 错误处理 - 如果无法识别增量字段，必须询问用户指定 - 如果无法识别登账字段，必须询问用户指定 - 如果SQL语法检查失败，修正语法错误 - 如果文件保存失败，检查目录权限 - 如果多表关联复杂，询问用户提供关联关系 ## 相关工具 - `glob`：查找业务表定义文件 - `read`：读取文件内容 - `grep`：搜索字段模式和注释 - `write`：保存生成的JSON和SQL文件 - `bash`：测试SQL语法 - `question`：询问用户字段信息和确认