第十五章：日志挖掘（LogMiner）

文档信息

适用版本：Oracle 12c / 18c / 19c / 21c
阅读时间：约 45 分钟
前置知识：第十四章闪回技术

15.1 LogMiner 概述

15.1.1 LogMiner 功能

┌─────────────────────────────────────────────────────────────────┐
│                      LogMiner 功能                                │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  LogMiner 用于分析重做日志和归档日志中的变更信息：                    │
│                                                                 │
│  主要用途：                                                      │
│  ┌───────────────────────────────────────────────────────────┐  │
│  │ 1. 审计数据库变更                                          │  │
│  │ 2. 追踪用户操作                                            │  │
│  │ 3. 恢复误删除数据                                          │  │
│  │ 4. 分析性能问题（定位执行慢的 SQL）                         │  │
│  │ 5. 同步测试数据库                                          │  │
│  │ 6. 确认特定变更（如谁在什么时候修改了某行）                  │  │
│  └───────────────────────────────────────────────────────────┘  │
│                                                                 │
│  LogMiner 读取两类日志：                                          │
│  • 在线重做日志文件                                              │
│  • 归档日志文件                                                  │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

15.1.2 前提条件

-- 1. 确保数据库处于归档模式（分析归档日志需要）
SELECT log_mode FROM v$database;
-- 应该返回 ARCHIVELOG

-- 2. 创建 LogMiner 需要的表空间和用户
-- sys 用户通常已有必要的权限

-- 3. 设置补充日志（用于追踪列变更）
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA (PRIMARY KEY) COLUMNS;
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA (UNIQUE) COLUMNS;

-- 4. 查看补充日志状态
SELECT SUPPLEMENTAL_LOG_DATA_PK, 
       SUPPLEMENTAL_LOG_DATA_UI, 
       SUPPLEMENTAL_LOG_DATA_ALL
FROM v$database;

15.2 LogMiner 配置

15.2.1 创建数据字典

-- 方式1：提取数据字典到重做日志
BEGIN
    DBMS_LOGMNR_D.BUILD(
        options => DBMS_LOGMNR_D.STORE_IN_REDO_LOG
    );
END;
/

-- 方式2：提取数据字典到文件系统
BEGIN
    DBMS_LOGMNR_D.BUILD(
        dictionary_filename => 'dictionary.ora',
        dictionary_location => '/u01/app/oracle/oradata/ORCL/',
        options => DBMS_LOGMNR_D.STORE_IN_FLAT_FILE
    );
END;
/

-- 查看数据字典文件
!ls -la /u01/app/oracle/oradata/ORCL/dictionary.ora

15.2.2 添加日志文件

-- 方式1：添加单个日志
BEGIN
    DBMS_LOGMNR.ADD_LOGFILE(
        logfilename => '/u01/app/oracle/arch/ORCL/arch_001.arc',
        options => DBMS_LOGMNR.NEW
    );
END;
/

-- 方式2：添加多个日志
BEGIN
    -- 添加第一个日志
    DBMS_LOGMNR.ADD_LOGFILE(
        logfilename => '/u01/app/oracle/arch/ORCL/arch_001.arc',
        options => DBMS_LOGMNR.NEW
    );
    
    -- 添加更多日志
    DBMS_LOGMNR.ADD_LOGFILE(
        logfilename => '/u01/app/oracle/arch/ORCL/arch_002.arc',
        options => DBMS_LOGMNR.ADDFILE
    );
END;
/

-- 方式3：自动发现所有归档日志
DECLARE
    v_log_files DBMS_APPLICATION_INFO.ATTR_CLOB;
BEGIN
    FOR rec IN (
        SELECT name FROM v$archived_log 
        WHERE name IS NOT NULL
        AND completion_time >= SYSDATE - 7
        ORDER BY sequence#
    ) LOOP
        DBMS_LOGMNR.ADD_LOGFILE(
            logfilename => rec.name,
            options => DBMS_LOGMNR.ADDFILE
        );
    END LOOP;
END;
/

15.3 启动 LogMiner 并分析

15.3.1 启动 LogMiner

-- 1. 基本启动
BEGIN
    DBMS_LOGMNR.START_LOGMNR(
        options => DBMS_LOGMNR.DICT_FROM_REDO_LOG
    );
END;
/

-- 2. 使用字典文件启动
BEGIN
    DBMS_LOGMNR.START_LOGMNR(
        dictfilename => '/u01/app/oracle/oradata/ORCL/dictionary.ora',
        starttime => TO_DATE('2024-01-15 09:00:00', 'YYYY-MM-DD HH24:MI:SS'),
        endtime => TO_DATE('2024-01-15 10:00:00', 'YYYY-MM-DD HH24:MI:SS')
    );
END;
/

-- 3. 基于 SCN 范围
BEGIN
    DBMS_LOGMNR.START_LOGMNR(
        startscn => 12345678,
        endscn => 12350000,
        dictfilename => '/u01/app/oracle/oradata/ORCL/dictionary.ora'
    );
END;
/

-- 4. 包含 SQL_REDO 和 SQL_UNDO
BEGIN
    DBMS_LOGMNR.START_LOGMNR(
        dictfilename => '/u01/app/oracle/oradata/ORCL/dictionary.ora',
        starttime => TO_DATE('2024-01-15 09:00:00', 'YYYY-MM-DD HH24:MI:SS'),
        endtime => TO_DATE('2024-01-15 10:00:00', 'YYYY-MM-DD HH24:MI:SS'),
        options => DBMS_LOGMNR.COMMITTED_DATA_ONLY + 
                   DBMS_LOGMNR.STRING_literals_in_DATA
    );
END;
/

-- 5. 关闭 LogMiner
BEGIN
    DBMS_LOGMNR.END_LOGMNR;
END;
/

15.3.2 LogMiner 选项

选项	说明
DICT_FROM_REDO_LOG	从重做日志读取字典
STORE_IN_FLAT_FILE	将字典存储到文件
STORE_IN_REDO_LOG	将字典存储到重做日志
COMMITTED_DATA_ONLY	只显示已提交事务
NO_ROWID_IN_STMT	SQL 语句中不包含 ROWID
STRING_LITERALS_IN_DATA	使用字符串格式显示数据

15.4 查询日志内容

15.4.1 基本查询

-- 查看所有变更
SELECT 
    timestamp,
    seg_owner,
    seg_name,
    operation,
    sql_redo,
    sql_undo,
    username
FROM v$logmnr_contents
ORDER BY timestamp;

-- 查看特定表的变更
SELECT 
    timestamp,
    operation,
    sql_redo,
    sql_undo,
    data_obj#,
    dataobj#,
    seg_owner,
    seg_name
FROM v$logmnr_contents
WHERE seg_name = 'EMPLOYEES'
AND seg_owner = 'HR'
ORDER BY timestamp;

-- 查看特定用户的操作
SELECT 
    timestamp,
    username,
    operation,
    sql_redo
FROM v$logmnr_contents
WHERE username = 'HR'
ORDER BY timestamp;

-- 查看 DML 操作
SELECT 
    timestamp,
    operation,
    seg_name,
    sql_redo
FROM v$logmnr_contents
WHERE operation IN ('INSERT', 'UPDATE', 'DELETE')
ORDER BY timestamp;

-- 查看 DDL 操作
SELECT 
    timestamp,
    operation,
    sql_redo,
    seg_name
FROM v$logmnr_contents
WHERE operation = 'DDL'
ORDER BY timestamp;

15.4.2 详细查询

-- 查看事务级别信息
SELECT 
    session#,
    serial#,
    username,
    transaction_name,
    COUNT(*) AS operation_count,
    MIN(timestamp) AS start_time,
    MAX(timestamp) AS end_time
FROM v$logmnr_contents
WHERE operation = 'INSERT'
GROUP BY session#, serial#, username, transaction_name
ORDER BY start_time DESC;

-- 查看特定 SCN 范围
SELECT 
    scn,
    timestamp,
    operation,
    sql_redo
FROM v$logmnr_contents
WHERE scn BETWEEN 12345678 AND 12350000
ORDER BY scn;

-- 查看回滚操作
SELECT 
    timestamp,
    operation,
    sql_undo
FROM v$logmnr_contents
WHERE operation = 'DELETE'
AND seg_name = 'EMPLOYEES'
ORDER BY timestamp;

-- 查看列值变化
SELECT 
    timestamp,
    operation,
    seg_name,
    DBMS_LOGMNR.COLUMN_PRESENT(
        row_data, 
        'SALARY'
    ) AS has_salary_change,
    row_data
FROM v$logmnr_contents
WHERE seg_name = 'EMPLOYEES'
AND operation = 'UPDATE';

15.5 实用案例

15.5.1 追踪误删除数据

-- 场景：某行数据被误删除，需要找回

-- 1. 启动 LogMiner（确定大概时间）
BEGIN
    DBMS_LOGMNR.START_LOGMNR(
        dictfilename => '/u01/app/oracle/oradata/ORCL/dictionary.ora',
        starttime => TO_DATE('2024-01-15 09:00:00', 'YYYY-MM-DD HH24:MI:SS'),
        endtime => TO_DATE('2024-01-15 12:00:00', 'YYYY-MM-DD HH24:MI:SS')
    );
END;
/

-- 2. 查找删除操作
SELECT 
    timestamp,
    operation,
    sql_undo,  -- 用于恢复的 INSERT 语句
    seg_name,
    session#,
    serial#
FROM v$logmnr_contents
WHERE operation = 'DELETE'
AND seg_name = 'EMPLOYEES'
AND seg_owner = 'HR'
AND sql_undo LIKE '%1000%'  -- 假设员工 ID 是 1000
ORDER BY timestamp;

-- 3. 执行 UNDO 语句恢复数据
-- 先查看 UNDO 内容
SELECT sql_undo FROM v$logmnr_contents
WHERE operation = 'DELETE'
AND seg_name = 'EMPLOYEES'
AND ROWNUM = 1;

-- 4. 手动执行恢复（复制 UNDO 内容执行）

-- 5. 结束 LogMiner
BEGIN
    DBMS_LOGMNR.END_LOGMNR;
END;
/

15.5.2 审计用户操作

-- 场景：审计 HR 用户的所有操作

-- 1. 分析日志
BEGIN
    DBMS_LOGMNR.START_LOGMNR(
        dictfilename => '/u01/app/oracle/oradata/ORCL/dictionary.ora',
        starttime => TO_DATE('2024-01-15 00:00:00', 'YYYY-MM-DD HH24:MI:SS'),
        endtime => TO_DATE('2024-01-15 23:59:59', 'YYYY-MM-DD HH24:MI:SS'),
        options => DBMS_LOGMNR.COMMITTED_DATA_ONLY
    );
END;
/

-- 2. 汇总用户操作
SELECT 
    username,
    operation,
    COUNT(*) AS operation_count
FROM v$logmnr_contents
WHERE username = 'HR'
GROUP BY username, operation
ORDER BY COUNT(*) DESC;

-- 3. 查看具体操作时间线
SELECT 
    TO_CHAR(timestamp, 'YYYY-MM-DD HH24:MI:SS') AS time,
    operation,
    seg_name,
    SUBSTR(sql_redo, 1, 200) AS sql_statement
FROM v$logmnr_contents
WHERE username = 'HR'
ORDER BY timestamp;

-- 4. 查看高风险操作（TRUNCATE、DROP）
SELECT 
    timestamp,
    operation,
    seg_name,
    sql_redo
FROM v$logmnr_contents
WHERE username = 'HR'
AND operation IN ('TRUNCATE', 'DROP')
ORDER BY timestamp;

BEGIN
    DBMS_LOGMNR.END_LOGMNR;
END;
/

15.5.3 同步测试数据库

-- 场景：将生产库的变更同步到测试库

-- 1. 分析需要的日志
BEGIN
    DBMS_LOGMNR.ADD_LOGFILE(
        logfilename => '/u01/app/oracle/arch/PROD/arch_001.arc',
        options => DBMS_LOGMNR.NEW
    );
    
    DBMS_LOGMNR.ADD_LOGFILE(
        logfilename => '/u01/app/oracle/arch/PROD/arch_002.arc',
        options => DBMS_LOGMNR.ADDFILE
    );
END;
/

-- 2. 启动 LogMiner
BEGIN
    DBMS_LOGMNR.START_LOGMNR(
        dictfilename => '/u01/app/oracle/oradata/PROD/dictionary.ora'
    );
END;
/

-- 3. 提取 SQL 语句
SELECT 
    'execute immediate ''' || REPLACE(sql_redo, '''', '''''') || ''';'
FROM v$logmnr_contents
WHERE operation IN ('INSERT', 'UPDATE', 'DELETE')
AND seg_owner = 'HR'
ORDER BY timestamp;

-- 4. 结束
BEGIN
    DBMS_LOGMNR.END_LOGMNR;
END;
/

15.5.4 定位问题 SQL

-- 场景：某时段性能下降，需要定位问题 SQL

-- 1. 启动 LogMiner
BEGIN
    DBMS_LOGMNR.START_LOGMNR(
        dictfilename => '/u01/app/oracle/oradata/ORCL/dictionary.ora',
        starttime => TO_DATE('2024-01-15 14:00:00', 'YYYY-MM-DD HH24:MI:SS'),
        endtime => TO_DATE('2024-01-15 15:00:00', 'YYYY-MM-DD HH24:MI:SS')
    );
END;
/

-- 2. 按操作类型统计
SELECT 
    operation,
    COUNT(*) AS count,
    SUM(LENGTH(sql_redo)) AS total_sql_length
FROM v$logmnr_contents
GROUP BY operation
ORDER BY count DESC;

-- 3. 找出最复杂的 UPDATE
SELECT 
    timestamp,
    seg_name,
    LENGTH(sql_redo) AS sql_length,
    SUBSTR(sql_redo, 1, 300) AS sql_preview
FROM v$logmnr_contents
WHERE operation = 'UPDATE'
ORDER BY LENGTH(sql_redo) DESC
FETCH FIRST 10 ROWS ONLY;

-- 4. 找出高频小操作
SELECT 
    username,
    seg_name,
    operation,
    COUNT(*) AS execution_count,
    MIN(sql_redo) AS sample_sql
FROM v$logmnr_contents
WHERE operation IN ('INSERT', 'UPDATE', 'DELETE')
GROUP BY username, seg_name, operation
HAVING COUNT(*) > 100
ORDER BY COUNT(*) DESC;

BEGIN
    DBMS_LOGMNR.END_LOGMNR;
END;
/

15.6 最佳实践

15.6.1 性能优化

-- 1. 限制分析范围
-- 尽量指定明确的 starttime/endtime 或 startscn/endscn

-- 2. 使用 COMMITTED_DATA_ONLY
-- 只显示已提交事务，减少数据量

-- 3. 定期清理日志
-- 旧日志分析价值降低，可以定期清理

-- 4. 使用 DBMS_LOGMNR.STORE_IN_REDO_LOG
-- 将字典信息存储在日志中，便于后续分析

-- 5. 大日志分析
-- 如果日志很大，可以分批分析
DECLARE
    v_start_scn NUMBER := 12345678;
    v_end_scn NUMBER;
    v_batch_size NUMBER := 10000;
BEGIN
    FOR i IN 1..100 LOOP
        v_end_scn := v_start_scn + v_batch_size;
        
        DBMS_LOGMNR.START_LOGMNR(
            startscn => v_start_scn,
            endscn => v_end_scn,
            dictfilename => '/u01/app/oracle/oradata/ORCL/dictionary.ora'
        );
        
        -- 分析操作...
        -- SELECT COUNT(*) FROM v$logmnr_contents;
        
        DBMS_LOGMNR.END_LOGMNR;
        
        v_start_scn := v_end_scn + 1;
    END LOOP;
END;
/

15.6.2 完整分析流程

-- 完整 LogMiner 分析流程
SET SERVEROUTPUT ON
DECLARE
    v_start_time DATE := TO_DATE('2024-01-15 09:00:00', 'YYYY-MM-DD HH24:MI:SS');
    v_end_time DATE := TO_DATE('2024-01-15 10:00:00', 'YYYY-MM-DD HH24:MI:SS');
    v_dict_loc VARCHAR2(100) := '/u01/app/oracle/oradata/ORCL/';
    v_dict_file VARCHAR2(100) := v_dict_loc || 'dictionary.ora';
    v_count NUMBER;
BEGIN
    -- Step 1: 创建字典文件
    DBMS_OUTPUT.PUT_LINE('Building dictionary...');
    DBMS_LOGMNR_D.BUILD(
        dictionary_filename => v_dict_file,
        dictionary_location => v_dict_loc,
        options => DBMS_LOGMNR_D.STORE_IN_FLAT_FILE
    );
    
    -- Step 2: 添加日志文件
    DBMS_OUTPUT.PUT_LINE('Adding log files...');
    FOR rec IN (
        SELECT name FROM v$archived_log 
        WHERE completion_time BETWEEN v_start_time AND v_end_time
        AND name IS NOT NULL
    ) LOOP
        IF SQL%ROWCOUNT = 0 THEN
            DBMS_LOGMNR.ADD_LOGFILE(rec.name, DBMS_LOGMNR.NEW);
        ELSE
            DBMS_LOGMNR.ADD_LOGFILE(rec.name, DBMS_LOGMNR.ADDFILE);
        END IF;
    END LOOP;
    
    -- Step 3: 启动 LogMiner
    DBMS_OUTPUT.PUT_LINE('Starting LogMiner...');
    DBMS_LOGMNR.START_LOGMNR(
        dictfilename => v_dict_file,
        starttime => v_start_time,
        endtime => v_end_time,
        options => DBMS_LOGMNR.DICT_FROM_REDO_LOG + 
                   DBMS_LOGMNR.COMMITTED_DATA_ONLY
    );
    
    -- Step 4: 分析操作
    DBMS_OUTPUT.PUT_LINE('Analyzing operations...');
    SELECT COUNT(*) INTO v_count FROM v$logmnr_contents;
    DBMS_OUTPUT.PUT_LINE('Total operations: ' || v_count);
    
    -- Step 5: 输出结果
    FOR rec IN (
        SELECT timestamp, username, operation, sql_redo 
        FROM v$logmnr_contents
        WHERE ROWNUM <= 100
        ORDER BY timestamp
    ) LOOP
        DBMS_OUTPUT.PUT_LINE(
            rec.timestamp || ' | ' || rec.username || 
            ' | ' || rec.operation || ' | ' || 
            SUBSTR(rec.sql_redo, 1, 100)
        );
    END LOOP;
    
    -- Step 6: 结束
    DBMS_LOGMNR.END_LOGMNR;
    DBMS_OUTPUT.PUT_LINE('LogMiner completed.');
    
EXCEPTION
    WHEN OTHERS THEN
        DBMS_LOGMNR.END_LOGMNR;
        RAISE;
END;
/

15.7 本章小结

✅ LogMiner 功能：分析重做日志和归档日志中的变更
✅ 前提条件：归档模式、补充日志
✅ 数据字典：创建和存储方式
✅ 日志添加：单个/多个日志，自动发现
✅ 启动选项：时间/SCN 范围，提交数据过滤
✅ 查询内容：sql_redo/sql_undo/操作类型
✅ 实用场景：误删除恢复、审计追踪、问题定位

📖 下章预告：高可用方案

下一步：学习第十六章：高可用方案