当前位置：首页 > news >正文

Doris与DS结合实现MySQL侧的Upsert功能

news 来源：原创 2025/6/14 11:48:34

可以借助 DolphinScheduler 调度工具，实现从 Doris 查询联合主键记录并传递到 MySQL 执行删除操作，需要结合 SQL 任务、Shell 任务和 参数传递 功能。

以下是具体步骤：

1. 创建工作流并添加 Doris 查询任务

首先添加一个 SQL 任务 查询 Doris 表中的联合主键：

任务类型：选择 SQL
数据源：选择已配置的 Doris 数据源

SQL 语句：

SELECT GROUP_CONCAT(CONCAT_WS(',', id1, id2) ,'|') AS primary_key  -- 联合主键字段用逗号连接
FROM doris_table
WHERE condition = 'value'

任务名称：例如 query_doris_pk
高级配置：勾选 将查询结果保存为变量，变量名设为 pk_list

语法解析

CONCAT_WS(',', id1, id2)：
- 将联合主键 id1 和 id2 用逗号 , 连接，例如 id1=1 和 id2=234 会转为 1,234。
GROUP_CONCAT(... SEPARATOR '|')：
- 将所有 CONCAT_WS 结果用竖线 | 连接，形成最终的一行字符串。
- 例如三条记录会转为 1,234|5,678|9,1011。

注意事项

NULL 值处理：
- CONCAT_WS 会忽略 NULL 值，若主键可能为 NULL，建议先使用 COALESCE 处理：
```
CONCAT_WS(',', COALESCE(id1, ''), COALESCE(id2, ''))
```
性能优化：
- 大数据量时，GROUP_CONCAT 可能导致内存问题，建议先过滤数据或分批处理。
字符集兼容：
- 确保 Doris 表的字符集与后续处理兼容（如 MySQL 的 UTF8 字符集）。

2. 添加 Shell 任务处理结果

添加一个 Shell 任务 处理查询结果，生成 MySQL 删除语句：

任务类型：选择 Shell

脚本内容：

# 读取 Doris 查询结果
PK_LIST="${pk_list}"# 生成 MySQL 删除语句
DELETE_SQL="DELETE FROM mysql_table WHERE (id1, id2) IN ("# 处理每条记录
IFS='|' read -ra RECORDS <<< "$PK_LIST"
for RECORD in "${RECORDS[@]}"; do# 分割联合主键IFS=',' read -ra KEYS <<< "$RECORD"DELETE_SQL+="('${KEYS[0]}','${KEYS[1]}'),"
done# 移除最后一个逗号并闭合括号
DELETE_SQL="${DELETE_SQL%,}"
DELETE_SQL+=");"# 输出结果供下游任务使用
echo "DELETE_SQL=${DELETE_SQL}"

任务名称：例如 generate_delete_sql
高级配置：勾选 将脚本输出作为变量，变量名设为 delete_sql

3. 添加 MySQL 执行任务

添加一个 SQL 任务 执行生成的删除语句：

任务类型：选择 SQL
数据源：选择已配置的 MySQL 数据源
SQL 语句：
```
${delete_sql}
```
任务名称：例如 execute_mysql_delete

4. 工作流配置与依赖关系

工作流参数：在工作流参数中定义变量：
```
pk_list = ""
delete_sql = ""
```

任务依赖：

query_doris_pk → generate_delete_sql → execute_mysql_delete

5. 验证与优化

测试查询：
- 先单独运行 query_doris_pk 任务，检查 pk_list 变量是否包含正确的联合主键。
SQL 安全：
- 如果主键值包含特殊字符，可能需要在 Shell 任务中增加转义处理。
批量处理：
- 对于大量数据，可将 pk_list 拆分为多个子任务，避免单条 SQL 过长。

完整工作流示例

+---------------------+      +-------------------------+      +------------------------+
| query_doris_pk      | ---> | generate_delete_sql     | ---> | execute_mysql_delete   |
| (SQL 任务)          |      | (Shell 任务)            |      | (SQL 任务)             |
| 从 Doris 查询主键    |      | 处理结果生成 SQL        |      | 执行 MySQL 删除        |
+---------------------+      +-------------------------+      +------------------------+