如何在Snowflake数据库中统计全表重复记录并构建数据质量监控表

方案选型建议 #

优先选择BASE表+定时刷新的方案，原因如下：

• 视图（VIEW）是实时计算，但多Schema多表的重复检测会涉及大量数据扫描，查询性能差，不适合日常监控场景
• BASE表存储历史检测结果，既能快速查看当前重复情况，也能追踪重复记录的变化趋势

步骤1：创建数据质量记录表 #

先构建一张用于存储重复记录检测结果的基表：

CREATE OR REPLACE TABLE DATA_QUALITY_DUPLICATES (
    SCHEMA_NAME VARCHAR(128),
    TABLE_NAME VARCHAR(128),
    DUPLICATE_C_VALUE VARCHAR(1000), -- 根据列C的实际类型调整（如INT/DECIMAL/DATE等）
    DUPLICATE_COUNT INT,
    DETECTION_TIMESTAMP TIMESTAMP DEFAULT CURRENT_TIMESTAMP()
);

注意：如果列C是数值型或日期型，请修改DUPLICATE_C_VALUE的字段类型以匹配实际数据类型。

步骤2：编写存储过程自动检测重复 #

创建JavaScript存储过程，遍历指定Schema下的所有表，检测列C的重复记录并插入到质量表：

CREATE OR REPLACE PROCEDURE DETECT_DUPLICATES_BY_C()
RETURNS VARCHAR
LANGUAGE JAVASCRIPT
EXECUTE AS CALLER
AS
$$
    // 替换成你的实际Schema列表
    const schemaList = ['SCHEMA_1', 'SCHEMA_2', 'SCHEMA_3', 'SCHEMA_4', 'SCHEMA_5'];
    let resultMsg = '';

    // 可选：清空历史结果（若需保留历史则注释此行）
    snowflake.execute({sqlText: "TRUNCATE TABLE DATA_QUALITY_DUPLICATES;"});

    for (let schema of schemaList) {
        // 获取当前Schema下的所有基表
        const getTablesStmt = `
            SELECT TABLE_NAME 
            FROM INFORMATION_SCHEMA.TABLES 
            WHERE TABLE_SCHEMA = '${schema}' 
              AND TABLE_TYPE = 'BASE TABLE';
        `;
        const tablesResult = snowflake.execute({sqlText: getTablesStmt});

        // 遍历每个表执行重复检测
        while (tablesResult.next()) {
            const tableName = tablesResult.getColumnValue(1);
            try {
                // 检查表是否包含列C
                const checkColStmt = `
                    SELECT COUNT(*) 
                    FROM INFORMATION_SCHEMA.COLUMNS 
                    WHERE TABLE_SCHEMA = '${schema}' 
                      AND TABLE_NAME = '${tableName}' 
                      AND COLUMN_NAME = 'C';
                `;
                const colCheckResult = snowflake.execute({sqlText: checkColStmt});
                colCheckResult.next();
                const colExists = colCheckResult.getColumnValue(1);

                if (colExists === 1) {
                    // 插入重复记录到质量表
                    const insertStmt = `
                        INSERT INTO DATA_QUALITY_DUPLICATES (SCHEMA_NAME, TABLE_NAME, DUPLICATE_C_VALUE, DUPLICATE_COUNT)
                        SELECT '${schema}', '${tableName}', C, COUNT(*) AS DUPLICATE_COUNT
                        FROM ${schema}.${tableName}
                        GROUP BY C
                        HAVING COUNT(*) > 1;
                    `;
                    snowflake.execute({sqlText: insertStmt});
                    resultMsg += `成功检测 ${schema}.${tableName} 的重复记录\n`;
                } else {
                    resultMsg += `${schema}.${tableName} 不存在列C，跳过检测\n`;
                }
            } catch (err) {
                resultMsg += `${schema}.${tableName} 检测失败：${err.message}\n`;
            }
        }
    }
    return resultMsg;
$$;

步骤3：执行存储过程并设置定时任务 #

手动执行测试 #

CALL DETECT_DUPLICATES_BY_C();

执行后可查询质量表查看结果：

SELECT * FROM DATA_QUALITY_DUPLICATES ORDER BY DETECTION_TIMESTAMP DESC;

设置定时自动检测 #

创建Snowflake任务，每天凌晨1点（UTC时间）自动执行检测：

CREATE OR REPLACE TASK DETECT_DUPLICATES_TASK
WAREHOUSE = YOUR_WAREHOUSE_NAME -- 替换成你的仓库名
SCHEDULE = 'USING CRON 0 1 * * * UTC' -- 可根据业务时区调整执行时间
AS
CALL DETECT_DUPLICATES_BY_C();

启动任务：

ALTER TASK DETECT_DUPLICATES_TASK RESUME;

注意事项 #

• 确保存储过程调用者拥有所有目标Schema表的SELECT权限，以及DATA_QUALITY_DUPLICATES表的INSERT/TRUNCATE权限
• 若需保留历史检测记录，删除存储过程中的TRUNCATE TABLE语句，每次检测会追加新结果
• 若列C是敏感数据，可对DUPLICATE_C_VALUE字段做脱敏处理
• 大表检测建议选择业务低峰期执行，避免影响正常业务

内容的提问来源于stack exchange，提问作者biggboss2019