DuckDB MERGE INTO 进阶：软删除、批量优化与生产环境最佳实践

深入 DuckDB MERGE INTO 的高级用法：软删除模式、大数据量分批处理、WAL 调优与生产环境最佳实践，附完整可运行代码。

DuckDB MERGE INTO 进阶：软删除、批量优化与生产环境最佳实践

难度：⭐⭐⭐ ｜ 预计时间：20 分钟掌握，从此告别脏数据

DuckDB MERGE INTO 生产环境架构

1. 为什么 MERGE INTO 是 ETL 工程师的瑞士军刀？

在数据工程领域，增量同步是最基础也最频繁的操作。每天你都要面对这样的场景：

用户表需要合并今天的变更（更新已有用户，插入新用户）
订单流水需要同步上游系统的状态变更
商品库存需要根据实时销售数据动态调整

传统的做法是用 Python 写循环判断——先查是否存在，再决定 UPDATE 还是 INSERT。代码冗长、容易出错、并发场景下还会产生竞态条件。

DuckDB 的 MERGE INTO 语句用一条 SQL 搞定所有逻辑，而且支持 UPDATE、INSERT、DELETE 三种操作。本文聚焦生产环境中的高级用法，涵盖软删除、批量处理和性能调优。

2. 基础回顾：MERGE INTO 的三要素

MERGE INTO 的核心结构由三个部分组成：

MERGE INTO target_table AS target
USING source_data AS source
ON target.key = source.key          -- 匹配条件
WHEN MATCHED THEN UPDATE SET ...     -- 匹配时执行
WHEN NOT MATCHED THEN INSERT ...     -- 不匹配时执行

2.1 基本示例：用户数据增量同步

-- 创建目标表（模拟历史用户数据）
CREATE TABLE users AS
SELECT * FROM (VALUES
    (1, '张三', 28, '北京'),
    (2, '李四', 35, '上海'),
    (3, '王五', 22, '广州'),
    (4, '赵六', 41, '深圳')
) AS t(user_id, name, age, city);

-- 创建源表（今日更新数据）
CREATE TABLE new_users AS
SELECT * FROM (VALUES
    (2, '李四', 36, '上海'),
    (3, '王五', 23, '杭州'),
    (5, '孙七', 30, '成都'),
    (6, '周八', 27, '武汉')
) AS t(user_id, name, age, city);

-- 执行 MERGE INTO
MERGE INTO users AS target
USING new_users AS source
ON target.user_id = source.user_id
WHEN MATCHED THEN UPDATE SET
    name = source.name,
    age = source.age,
    city = source.city
WHEN NOT MATCHED THEN INSERT VALUES
    (source.user_id, source.name, source.age, source.city);

-- 查看结果
SELECT * FROM users ORDER BY user_id;

输出：

user_id | name  | age | city
--------|-------|-----|--------
1       | 张三  | 28  | 北京
2       | 李四  | 36  | 上海      ← 年龄更新 35→36
3       | 王五  | 23  | 杭州      ← 城市更新 广州→杭州
4       | 赵六  | 41  | 深圳      ← 保持不变
5       | 孙七  | 30  | 成都      ← 新增
6       | 周八  | 27  | 武汉      ← 新增

3. 进阶技巧一：软删除模式

生产环境中，一个常见需求是：当源数据中某个记录消失了，如何处理目标表中的对应记录？

硬删除（直接 DELETE）可能导致数据丢失且不可恢复。更优雅的做法是使用软删除——标记为已删除而非物理删除。

3.1 方案一：MERGE INTO 配合 deleted_users 表

-- 创建已删除用户标记表
CREATE TABLE deleted_users AS
SELECT * FROM (VALUES
    (4, '赵六')  -- 赵六今天被标记为离职
) AS t(user_id, reason);

-- MERGE INTO 同时处理更新、插入和删除
MERGE INTO users AS target
USING new_users AS source
ON target.user_id = source.user_id
WHEN MATCHED THEN UPDATE SET
    name = source.name,
    age = source.age,
    city = source.city
WHEN NOT MATCHED AND source.user_id IS NOT NULL THEN
    INSERT VALUES (source.user_id, source.name, source.age, source.city)
WHEN NOT MATCHED AND target.user_id IN (SELECT user_id FROM deleted_users) THEN
    UPDATE SET is_deleted = true;

3.2 方案二：添加 is_deleted 字段（推荐）

更通用的做法是在表中预留 is_deleted 和 deleted_at 字段：

-- 重建带软删除字段的表
DROP TABLE IF EXISTS users;
CREATE TABLE users (
    user_id INTEGER PRIMARY KEY,
    name VARCHAR,
    age INTEGER,
    city VARCHAR,
    is_deleted BOOLEAN DEFAULT false,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- 插入初始数据
INSERT INTO users VALUES
    (1, '张三', 28, '北京', false, '2026-07-13'),
    (2, '李四', 35, '上海', false, '2026-07-13'),
    (3, '王五', 22, '广州', false, '2026-07-13'),
    (4, '赵六', 41, '深圳', false, '2026-07-13');

-- 执行 MERGE INTO，自动标记消失的用户
MERGE INTO users AS target
USING new_users AS source
ON target.user_id = source.user_id
WHEN MATCHED THEN UPDATE SET
    name = source.name,
    age = source.age,
    city = source.city,
    updated_at = CURRENT_TIMESTAMP
WHEN NOT MATCHED THEN INSERT VALUES
    (source.user_id, source.name, source.age, source.city, false, CURRENT_TIMESTAMP)
WHEN NOT MATCHED THEN UPDATE SET
    is_deleted = true,
    updated_at = CURRENT_TIMESTAMP;

执行后，未出现在 new_users 中的用户（1号张三和4号赵六）会被标记为已删除。

3.3 查询时过滤已删除记录

-- 只查询活跃用户
SELECT * FROM users WHERE is_deleted = false ORDER BY user_id;

-- 查看被删除的用户
SELECT * FROM users WHERE is_deleted = true;

4. 进阶技巧二：大数据量分批处理

当源数据达到百万级甚至千万级时，一次性 MERGE INTO 可能导致内存溢出或事务超时。以下是三种优化策略：

4.1 按时间窗口过滤

只同步最近变更的数据，避免全量比对：

MERGE INTO users AS target
USING (
    SELECT * FROM new_users
    WHERE updated_at > (SELECT MAX(updated_at) FROM users)
) AS source
ON target.user_id = source.user_id
WHEN MATCHED THEN UPDATE SET
    name = source.name, age = source.age, city = source.city,
    updated_at = source.updated_at
WHEN NOT MATCHED THEN INSERT VALUES
    (source.user_id, source.name, source.age, source.city, false, source.updated_at);

4.2 分批 MERGE

将大表拆分为小批次逐个处理：

-- Python 中控制批次大小
batch_size = 50000

for offset in range(0, total_rows, batch_size):
    con.execute(f"""
        MERGE INTO users AS target
        USING (
            SELECT * FROM new_users
            LIMIT {batch_size} OFFSET {offset}
        ) AS source
        ON target.user_id = source.user_id
        WHEN MATCHED THEN UPDATE SET
            name = source.name, age = source.age, city = source.city
        WHEN NOT MATCHED THEN INSERT VALUES
            (source.user_id, source.name, source.age, source.city, false, CURRENT_TIMESTAMP)
    """)
    print(f"Processed batch at offset {offset}")

4.3 使用临时表预过滤

对于复杂的过滤逻辑，先用临时表缩小数据范围：

-- 步骤1：创建临时过滤表
CREATE TEMP TABLE temp_merge_source AS
SELECT * FROM new_users
WHERE updated_at >= DATE('now', '-1 day')
AND status = 'active';

-- 步骤2：对临时表执行 MERGE
MERGE INTO users AS target
USING temp_merge_source AS source
ON target.user_id = source.user_id
WHEN MATCHED THEN UPDATE SET ...
WHEN NOT MATCHED THEN INSERT ...;

-- 步骤3：清理临时表
DROP TABLE temp_merge_source;

5. 进阶技巧三：性能调优

5.1 关闭 WAL 提升写入速度

DuckDB 默认开启 WAL（Write-Ahead Logging），在内存数据库场景中可以关闭以提升写入性能：

import duckdb

con = duckdb.connect(":memory:")
con.execute("PRAGMA wal_off;")

# 执行 MERGE INTO
con.execute("""
    MERGE INTO users AS target
    USING new_users AS source
    ON target.user_id = source.user_id
    WHEN MATCHED THEN UPDATE SET name = source.name
    WHEN NOT MATCHED THEN INSERT VALUES (source.user_id, source.name)
""")

5.2 调整并行度

-- 设置最大并行线程数
PRAGMA threads=8;

-- 查看当前配置
PRAGMA threads;
PRAGMA memory_limit;

5.3 创建索引加速匹配

虽然 DuckDB 的列式存储在扫描场景下表现优异，但对于 MERGE INTO 的匹配环节，主键或唯一索引可以显著加速 ON 条件的查找：

-- 为目标表创建索引
CREATE INDEX idx_users_user_id ON users(user_id);

-- 为源表创建索引
CREATE INDEX idx_new_users_user_id ON new_users(user_id);

-- 执行 MERGE INTO
MERGE INTO users AS target
USING new_users AS source
ON target.user_id = source.user_id
WHEN MATCHED THEN UPDATE SET ...
WHEN NOT MATCHED THEN INSERT ...;

5.4 使用 EXPLAIN ANALYZE 诊断性能

EXPLAIN ANALYZE
MERGE INTO users AS target
USING new_users AS source
ON target.user_id = source.user_id
WHEN MATCHED THEN UPDATE SET
    name = source.name, age = source.age, city = source.city
WHEN NOT MATCHED THEN INSERT VALUES
    (source.user_id, source.name, source.age, source.city);

输出会显示每个阶段的耗时，帮助你定位瓶颈：

Merge Into
  Output: target.user_id, target.name, target.age, target.city
  -> Hash Join (Inner, Left Anti)
       Hash Cond: target.user_id = source.user_id
       -> Table Scan on users
       -> Table Scan on new_users

6. 实战：Python ETL 完整示例

以下是一个完整的 Python ETL 脚本，整合了上述所有技巧：

import duckdb
import pandas as pd
from datetime import datetime

class IncrementalETL:
    def __init__(self, db_path="data.db"):
        self.con = duckdb.connect(db_path)
        # 关闭 WAL 提升写入性能
        self.con.execute("PRAGMA wal_off;")
        # 设置并行度
        self.con.execute("PRAGMA threads=4;")
    
    def init_tables(self):
        """初始化目标表和源表"""
        self.con.execute("""
            CREATE TABLE IF NOT EXISTS users (
                user_id INTEGER PRIMARY KEY,
                name VARCHAR,
                age INTEGER,
                city VARCHAR,
                is_deleted BOOLEAN DEFAULT false,
                updated_at TIMESTAMP
            )
        """)
    
    def merge_incremental(self, source_df, batch_size=50000):
        """
        执行增量 MERGE
        
        Args:
            source_df: 包含更新数据的 DataFrame
            batch_size: 每批处理的行数
        """
        # 注册为临时表
        self.con.register('new_users', source_df)
        
        # 创建索引加速匹配
        self.con.execute("CREATE INDEX IF NOT EXISTS idx_users_uid ON users(user_id)")
        self.con.execute("CREATE INDEX IF NOT EXISTS idx_new_users_uid ON new_users(user_id)")
        
        # 执行 MERGE INTO
        self.con.execute("""
            MERGE INTO users AS target
            USING new_users AS source
            ON target.user_id = source.user_id
            WHEN MATCHED THEN UPDATE SET
                name = source.name,
                age = source.age,
                city = source.city,
                is_deleted = false,
                updated_at = CURRENT_TIMESTAMP
            WHEN NOT MATCHED THEN INSERT VALUES
                (source.user_id, source.name, source.age, source.city, false, CURRENT_TIMESTAMP)
            WHEN NOT MATCHED THEN UPDATE SET
                is_deleted = true,
                updated_at = CURRENT_TIMESTAMP
        """)
        
        print(f"MERGE completed at {datetime.now()}")
    
    def get_active_users(self):
        """获取活跃用户列表"""
        return self.con.execute("""
            SELECT * FROM users 
            WHERE is_deleted = false 
            ORDER BY updated_at DESC
        """).fetchdf()
    
    def close(self):
        self.con.close()

# 使用示例
if __name__ == "__main__":
    etl = IncrementalETL("production.db")
    etl.init_tables()
    
    # 模拟每日更新数据
    new_data = pd.DataFrame({
        'user_id': [2, 3, 5, 6],
        'name': ['李四', '王五', '孙七', '周八'],
        'age': [36, 23, 30, 27],
        'city': ['上海', '杭州', '成都', '武汉']
    })
    
    etl.merge_incremental(new_data)
    
    # 查看结果
    active = etl.get_active_users()
    print(active)
    
    etl.close()

7. 与传统工具对比

特性	DuckDB MERGE INTO	Python 循环	PostgreSQL upsert	Spark MERGE
代码行数	1 条 SQL	10-20 行	1 条 SQL	需 DataFrame API
并发安全	✅ 内置事务	❌ 需额外锁	✅	✅
支持软删除	✅	⚠️ 手动实现	⚠️ 手动实现	⚠️ 手动实现
大数据量性能	⭐⭐⭐⭐	⭐⭐	⭐⭐⭐	⭐⭐⭐⭐
学习成本	低	中	中	高
部署复杂度	零	需 Python 环境	需 PG 服务	需 Spark 集群

8. 变现建议

掌握 MERGE INTO 的生产级用法，你可以：

搭建数据产品后端：用 DuckDB + FastAPI 构建实时更新的数据看板，为中小企业提供 SaaS 数据服务（月费 500-3000 元/客户）
自动化报表服务：为电商客户提供每日自动增量同步的报表系统，按数据量收费
数据清洗外包：很多传统企业有大量手工 Excel 数据需要清洗入库，MERGE INTO 可以让你的交付效率提升 10 倍
微服务数据同步：在 Go/Node.js 微服务架构中，用 DuckDB 做本地缓存层的增量同步，减少数据库压力

💡 更多 DuckDB 生产环境最佳实践和 ETL 实战案例 → duckdblab.org

📺 Watch video tutorials → Olap Studio YouTube

Subscribe for more DuckDB & AI automation tutorials

⚠️ 本站为独立社区项目，与 DuckDB 基金会及 DuckDB 官方项目无任何从属、背书或赞助关系。

"DuckDB" 是 DuckDB 基金会的注册商标，本站仅以事实描述方式使用该名称。

本站内容仅供教育与社区推广用途，不构成任何商业服务。