Oracle LogMiner 日志挖掘：原理、步骤、可执行脚本与结果解析 – 刘子毅

1. LogMiner の原理

1.1 エッセンス

Oracle の REDO ログ レコード物理的変化ベクトル (変化ベクトル): データ ブロック アドレス (RDBA)、オペレーション コード、カラム オフセット、フロントミラーとバック ミラー、SCN、トランザクション ID (XID) などは、それ自体がバイナリであり、人間が読み取ることはできません。

LogMiner が行うことは、「物理 → 論理」の翻訳：

Redo/归档日志（二进制 change vector）
        │  ① 按 SCN / 事务顺序读取
        │  ② 用数据字典把 OBJ# / COL# 翻译成 表名 / 列名
        │  ③ 按事务重组、拼接行变更
        ▼
V$LOGMNR_CONTENTS（可读的逻辑 SQL）
        ├── SQL_REDO  → 原始操作（重放用，如 DELETE FROM T WHERE ...）
        └── SQL_UNDO  → 反向操作（回滚用，如 INSERT INTO T VALUES (...)）

1.2 必要な3つの要素

1.3 主要なメカニズムと制限事項

• 補足ログを有効にする必要があります: アーカイブを開くだけで、開かない SUPPLEMENTAL LOG DATA DML レコードが失われます (たとえば、10 レコードを削除した後、6 レコードのみが見つかります)。 SQL_UNDO NULL の可能性があります。最小限の補足ログで十分であり、主キー補足ログにより、 SQL_UNDO WHERE 条件の方がより正確です。
• DDL は参照のみ可能です SQL_REDO、いいえ SQL_UNDODDL ロールバックには手動処理が必要です。
• サポートされていない/不完全: 一部のLOB、LONG、XMLType、ADT、チェーン行/移行行のDMLが可能 SQL_REDO/SQL_UNDO NULL または Unsupported。
• クロスデータベース分析: 解析ライブラリ DB_BLOCK_SIZE 必須 ≥ ソース ライブラリ、プラットフォーム、文字セットが一貫している必要があり、使用する必要がありますソース生成された辞書。
• 12c 以降の CDB:LogMiner が必要です CDB$ROOT PDB で実行すると、PDB 内で実行するとエラーが報告されます。
• 現在のオンライン ログをマイニング リストに追加しないでくださいそれ以外の場合は CTAS/クエリ V$LOGMNR_CONTENTS トリガーしやすい ORA-00310 / ORA-00334(再帰的やり直しによりログの切り替えが発生し、問題が繰り返し発生します)。

2. 標準手順とコマンド（マニュアル版）

-- 0. 环境检查
ARCHIVE LOG LIST;
SELECT supplemental_log_data_min, supplemental_log_data_pk FROM v$database;

-- 1. 开启补充日志（只需做一次，持续生效）
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA (PRIMARY KEY) COLUMNS;   -- 可选，回滚更精准

-- 2. 定位误操作时间窗内的归档日志
SELECT thread#, sequence#, first_time, next_time, name
  FROM v$archived_log
 WHERE name IS NOT NULL AND status="A"
   AND next_time  > TO_DATE('2026-10-23 10:00:00','YYYY-MM-DD HH24:MI:SS')
   AND first_time < TO_DATE('2026-10-23 10:30:00','YYYY-MM-DD HH24:MI:SS')
 ORDER BY thread#, sequence#;

-- 3. 添加日志（第一个用 NEW，后续用 ADDFILE）
EXEC DBMS_LOGMNR.ADD_LOGFILE('/arch/.../1_10506.dbf', DBMS_LOGMNR.NEW);
EXEC DBMS_LOGMNR.ADD_LOGFILE('/arch/.../1_10507.dbf', DBMS_LOGMNR.ADDFILE);

-- 4. 启动挖掘（同一会话内执行）
EXEC DBMS_LOGMNR.START_LOGMNR( -
       STARTTIME => TO_DATE('2026-10-23 10:00:00','YYYY-MM-DD HH24:MI:SS'), -
       ENDTIME   => TO_DATE('2026-10-23 10:30:00','YYYY-MM-DD HH24:MI:SS'), -
       OPTIONS   => DBMS_LOGMNR.DICT_FROM_ONLINE_CATALOG + -
                    DBMS_LOGMNR.COMMITTED_DATA_ONLY + -
                    DBMS_LOGMNR.PRINT_PRETTY_SQL + -
                    DBMS_LOGMNR.SKIP_CORRUPTION);

-- 5. 查询结果（务必带过滤条件，否则极慢）
SELECT scn, timestamp, xid, username, operation, seg_owner, seg_name,
       sql_redo, sql_undo
  FROM v$logmnr_contents
 WHERE seg_owner="SCOTT" AND seg_name="EMP"
   AND operation IN ('INSERT','UPDATE','DELETE');

-- 6. 结束挖掘，释放 PGA
EXEC DBMS_LOGMNR.END_LOGMNR;

よく使われる START_LOGMNR オプション:

許可要件:LOGMINING（12c+）、SELECT ANY TRANSACTION、EXECUTE_CATALOG_ROLEまたは直接 sysdba。

3. 実行可能スクリプト (コピーして名前を付けて保存) logminer_mine.sh）

スクリプトは自動的に完了します: 環境チェック → 時間枠に従ってアーカイブされたログを自動的に収集 → マイニングを追加して開始 → 結果がテーブルにドロップされる → 多次元統計分析 → 自動スプライシングによるロールバック スクリプトの生成(CSF処理継続) → マイニングを終了します。

#!/usr/bin/env bash
###############################################################################
# logminer_mine.sh —— Oracle LogMiner 归档日志挖掘 + 结果解析 + 回滚脚本生成
# 适用：Oracle 11gR2 / 12c / 18c / 19c / 21c（CDB 请在 CDB$ROOT 执行）
# 用法：
#   ./logminer_mine.sh "user/pass@tns" "2026-10-23 10:00:00" "2026-10-23 10:30:00"
#   ./logminer_mine.sh "/ as sysdba"  "2026-10-23 10:00:00" "2026-10-23 10:30:00" "SCOTT.EMP"
# 参数：
#   $1 连接串   $2 起始时间   $3 结束时间
#   $4 可选：对象过滤 OWNER.TABLE
#   $5 可选：输出目录（默认 $HOME/logminer_out）
###############################################################################
set -euo pipefail

CONN="${1:?用法: $0 \"user/pass@tns\" \"开始时间\" \"结束时间\" [OWNER.TABLE] [输出目录]}"
START_TIME="${2:?缺少开始时间}"
END_TIME="${3:?缺少结束时间}"
OBJ="${4:-}"
OUT_DIR="${5:-$HOME/logminer_out}"

mkdir -p "$OUT_DIR"
TS=$(date +%Y%m%d_%H%M%S)
STAGE="LOGMNR_STAGE_${TS}"
UNDO_TAB="LOGMNR_UNDO_${TS}"
LOG_FILE="$OUT_DIR/logminer_${TS}.log"
REPORT_FILE="$OUT_DIR/report_${TS}.txt"
UNDO_FILE="$OUT_DIR/undo_${TS}.sql"

# 对象过滤条件
OBJ_WHERE=""
if [ -n "$OBJ" ]; then
  case "$OBJ" in
    *.*) O="${OBJ%%.*}"; T="${OBJ##*.}" ;;
    *)   O="";           T="$OBJ" ;;
  esac
  if [ -n "$O" ]; then
    OBJ_WHERE=" AND SEG_OWNER='${O^^}' AND SEG_NAME='${T^^}'"
  else
    OBJ_WHERE=" AND SEG_NAME='${T^^}'"
  fi
fi

echo "[INFO] 时间窗 : ${START_TIME} ~ ${END_TIME}"
echo "[INFO] 对象过滤: ${OBJ:-<全部用户对象>}"
echo "[INFO] 输出目录: ${OUT_DIR}"

sqlplus -s "$CONN" < "$LOG_FILE" 2>&1
SET ECHO OFF
SET FEEDBACK ON
SET VERIFY OFF
SET DEFINE OFF
SET PAGESIZE 300
SET LINESIZE 400
SET TRIMSPOOL ON
SET LONG 1000000
SET LONGCHUNKSIZE 1000000
SET SERVEROUTPUT ON SIZE UNLIMITED FORMAT WRAPPED
WHENEVER SQLERROR EXIT FAILURE

ALTER SESSION SET NLS_DATE_FORMAT='YYYY-MM-DD HH24:MI:SS';
ALTER SESSION SET NLS_DATE_LANGUAGE='AMERICAN';

SPOOL "${REPORT_FILE}"

PROMPT ################ 1. 环境检查 ################
SELECT NAME AS DB_NAME, LOG_MODE, DATABASE_ROLE, CDB,
       SUPPLEMENTAL_LOG_DATA_MIN AS SUP_MIN,
       SUPPLEMENTAL_LOG_DATA_PK  AS SUP_PK
  FROM V\$DATABASE;

SELECT OWNER, OBJECT_NAME, OBJECT_TYPE, STATUS
  FROM DBA_OBJECTS
 WHERE OBJECT_NAME IN ('DBMS_LOGMNR','DBMS_LOGMNR_D')
   AND OWNER = 'SYS';

PROMPT ################ 2. 待分析归档日志 ################
SELECT THREAD#, SEQUENCE#, FIRST_TIME, NEXT_TIME, NAME
  FROM V\$ARCHIVED_LOG
 WHERE NAME IS NOT NULL AND STATUS = 'A'
   AND NEXT_TIME  > TO_DATE('${START_TIME}','YYYY-MM-DD HH24:MI:SS')
   AND FIRST_TIME < TO_DATE('${END_TIME}','YYYY-MM-DD HH24:MI:SS')
 ORDER BY THREAD#, SEQUENCE#;

PROMPT ################ 3. 添加日志并启动挖掘 ################
DECLARE
  v_first BOOLEAN := TRUE;
  v_cnt   PLS_INTEGER := 0;
BEGIN
  FOR r IN (
    SELECT DISTINCT name, thread#, sequence#
      FROM V\$ARCHIVED_LOG
     WHERE NAME IS NOT NULL AND STATUS = 'A'
       AND NEXT_TIME  > TO_DATE('${START_TIME}','YYYY-MM-DD HH24:MI:SS')
       AND FIRST_TIME < TO_DATE('${END_TIME}','YYYY-MM-DD HH24:MI:SS')
     ORDER BY thread#, sequence#
  ) LOOP
    BEGIN
      IF v_first THEN
        DBMS_LOGMNR.ADD_LOGFILE(LOGFILENAME => r.name, OPTIONS => DBMS_LOGMNR.NEW);
        v_first := FALSE;
      ELSE
        DBMS_LOGMNR.ADD_LOGFILE(LOGFILENAME => r.name, OPTIONS => DBMS_LOGMNR.ADDFILE);
      END IF;
      v_cnt := v_cnt + 1;
    EXCEPTION
      WHEN OTHERS THEN
        DBMS_OUTPUT.PUT_LINE('跳过 ' || r.name || ' : ' || SQLERRM);
    END;
  END LOOP;

  IF v_cnt = 0 THEN
    RAISE_APPLICATION_ERROR(-20001, '指定时间窗内没有可用归档日志，请检查归档保留策略或时间范围');
  END IF;
  DBMS_OUTPUT.PUT_LINE('已添加归档日志数量: ' || v_cnt);
END;
/

BEGIN
  DBMS_LOGMNR.START_LOGMNR(
    STARTTIME => TO_DATE('${START_TIME}','YYYY-MM-DD HH24:MI:SS'),
    ENDTIME   => TO_DATE('${END_TIME}','YYYY-MM-DD HH24:MI:SS'),
    OPTIONS   => DBMS_LOGMNR.DICT_FROM_ONLINE_CATALOG
               + DBMS_LOGMNR.COMMITTED_DATA_ONLY
               + DBMS_LOGMNR.PRINT_PRETTY_SQL
               + DBMS_LOGMNR.SKIP_CORRUPTION
  );
  DBMS_OUTPUT.PUT_LINE('START_LOGMNR 完成');
END;
/

PROMPT ---- 当前分析参数 ----
SELECT START_SCN, START_DATE, END_SCN, END_DATE, OPTIONS
  FROM V\$LOGMNR_PARAMETERS;

PROMPT ---- 已注册日志清单 ----
SELECT THREAD#, SEQUENCE#, LOW_SCN, NEXT_SCN, FILENAME
  FROM V\$LOGMNR_LOGS
 ORDER BY THREAD#, SEQUENCE#;

PROMPT ################ 4. 结果落地（同一会话内） ################
CREATE TABLE ${STAGE} AS
SELECT SCN,
       COMMIT_SCN,
       "TIMESTAMP"      AS CHANGE_TIME,
       START_TIMESTAMP  AS START_TIME,
       XID,
       USERNAME,
       SESSION#         AS SESSION_ID,
       OPERATION,
       SEG_OWNER,
       SEG_NAME,
       SEG_TYPE_NAME,
       SQL_REDO,
       SQL_UNDO,
       CSF,
       RS_ID,
       SSN,
       ROLLBACK         AS IS_ROLLBACK,
       SESSION_INFO,
       THREAD#          AS THREAD_ID,
       ROW_ID
  FROM V\$LOGMNR_CONTENTS
 WHERE OPERATION IN ('INSERT','UPDATE','DELETE','DDL')
   AND (SEG_OWNER IS NULL OR SEG_OWNER NOT IN
        ('SYS','SYSTEM','SYSMAN','DBSNMP','OUTLN','AUDSYS','XDB','MDSYS',
         'CTXSYS','ORDSYS','OLAPSYS','WMSYS','LBACSYS','APPQOSSYS',
         'GSMADMIN_INTERNAL','DVSYS','DVF','OJVMSYS','PUBLIC'))
${OBJ_WHERE};

BEGIN
  DBMS_LOGMNR.END_LOGMNR;
  DBMS_OUTPUT.PUT_LINE('END_LOGMNR 完成，PGA 已释放');
END;
/

PROMPT ################ 5. 挖掘结果解析 ################
PROMPT ---- 5.1 总览 ----
SELECT COUNT(*) AS TOTAL_ROWS,
       COUNT(DISTINCT XID) AS TX_COUNT,
       MIN(CHANGE_TIME) AS MIN_TIME,
       MAX(CHANGE_TIME) AS MAX_TIME
  FROM ${STAGE};

PROMPT ---- 5.2 按操作类型 ----
SELECT OPERATION, COUNT(*) AS CNT
  FROM ${STAGE}
 GROUP BY OPERATION
 ORDER BY CNT DESC;

PROMPT ---- 5.3 按对象 TOP 20 ----
SELECT * FROM (
  SELECT SEG_OWNER, SEG_NAME, OPERATION, COUNT(*) AS CNT
    FROM ${STAGE}
   GROUP BY SEG_OWNER, SEG_NAME, OPERATION
   ORDER BY CNT DESC
) WHERE ROWNUM <= 20;

PROMPT ---- 5.4 按数据库用户 ----
SELECT USERNAME, COUNT(*) AS CNT
  FROM ${STAGE}
 GROUP BY USERNAME
 ORDER BY CNT DESC;

PROMPT ---- 5.5 按小时分布 ----
SELECT TO_CHAR(CHANGE_TIME,'YYYY-MM-DD HH24') AS HOUR,
       OPERATION, COUNT(*) AS CNT
  FROM ${STAGE}
 GROUP BY TO_CHAR(CHANGE_TIME,'YYYY-MM-DD HH24'), OPERATION
 ORDER BY 1, 2;

PROMPT ---- 5.6 事务清单（按提交顺序）----
SELECT XID, USERNAME, COUNT(*) AS ROW_CNT,
       MIN(CHANGE_TIME) AS FIRST_TIME,
       MAX(CHANGE_TIME) AS LAST_TIME,
       MAX(COMMIT_SCN)  AS COMMIT_SCN
  FROM ${STAGE}
 GROUP BY XID, USERNAME
 ORDER BY COMMIT_SCN;

PROMPT ---- 5.7 变更明细（前 200 条）----
SELECT SCN, CHANGE_TIME, XID, USERNAME, OPERATION,
       SEG_OWNER, SEG_NAME, SQL_REDO
  FROM (SELECT * FROM ${STAGE} ORDER BY SCN, RS_ID, SSN)
 WHERE ROWNUM <= 200;

PROMPT ################ 6. 生成回滚脚本 ################
CREATE TABLE ${UNDO_TAB} (SEQ NUMBER, STMT CLOB);

DECLARE
  v_buf  CLOB;
  v_seq  PLS_INTEGER := 0;
  v_cnt  PLS_INTEGER := 0;
BEGIN
  FOR r IN (
    SELECT SQL_UNDO, CSF
      FROM ${STAGE}
     WHERE SQL_UNDO IS NOT NULL
       AND OPERATION IN ('INSERT','UPDATE','DELETE')
     ORDER BY COMMIT_SCN DESC, RS_ID DESC, SSN DESC   -- 逆序回滚
  ) LOOP
    IF v_buf IS NULL THEN
      v_buf := r.SQL_UNDO;
    ELSE
      v_buf := v_buf || r.SQL_UNDO;                    -- 拼接 CSF 续行
    END IF;

    IF NVL(r.CSF, 0) = 0 THEN
      v_seq := v_seq + 1;
      INSERT INTO ${UNDO_TAB}(SEQ, STMT)
      VALUES (v_seq,
              v_buf || CASE WHEN SUBSTR(v_buf, -1, 1) = ';' THEN '' ELSE ';' END);
      v_buf := NULL;
      v_cnt := v_cnt + 1;
    END IF;
  END LOOP;
  COMMIT;
  DBMS_OUTPUT.PUT_LINE('生成回滚语句条数: ' || v_cnt);
END;
/

SPOOL OFF

SET PAGESIZE 0
SET HEADING OFF
SET FEEDBACK OFF
SPOOL "${UNDO_FILE}"
PROMPT -- ==========================================================
PROMPT -- LogMiner 回滚脚本（请务必完整审核后再执行）
PROMPT -- 生成时间: ${TS}
PROMPT -- 源时间窗: ${START_TIME} ~ ${END_TIME}
PROMPT -- 建议: 先备份相关表；SET AUTOCOMMIT OFF；分批执行并校验
PROMPT -- 注意: DDL 无 SQL_UNDO，需人工处理
PROMPT -- ==========================================================
PROMPT
SELECT STMT FROM ${UNDO_TAB} ORDER BY SEQ;
SPOOL OFF

PROMPT ################ 7. 完成 ################
PROMPT 明细表: ${STAGE}
PROMPT 回滚表: ${UNDO_TAB}
PROMPT 回滚脚本: ${UNDO_FILE}
PROMPT 报告: ${REPORT_FILE}
SQLEOF

echo "[OK] 日志: $LOG_FILE"
echo "[OK] 报告: $REPORT_FILE"
echo "[OK] 回滚脚本: $UNDO_FILE"
echo "[HINT] 明细表 ${STAGE} 保留用于复核，确认后可手工 DROP"

使用例

chmod +x logminer_mine.sh

# 全库挖掘
./logminer_mine.sh "sys/Oracle_123@orcl" "2026-10-23 10:00:00" "2026-10-23 10:30:00"

# 只挖某张表
./logminer_mine.sh "/ as sysdba" "2026-10-23 10:00:00" "2026-10-23 10:30:00" "SCOTT.EMP"

4. マイニング後のデータ分析

4.1 V$LOGMNR_CONTENTS コア分野の解釈

4.2 解釈の順序（推奨）

1. 概要を見る：TOTAL_ROWS、TX_COUNT時間範囲が誤操作ウィンドウ内に収まるかどうか。
   • 0 の場合 → タイム ウィンドウが間違っているか、アーカイブが見つからないか、補足ログが開かれていないため、レコードが不完全になります。
2. ビューオブジェクトTOP: 影響を受けるテーブルがターゲット テーブルであるかどうかを確認します。予期しないテーブルが多数表示される場合は、時間枠が広すぎることを意味します。
3. 営業リストを見てみる：現れる LAST_TIME 誤操作のタイミングに合わせて、ROW_CNT 影響を受ける行の数と一致する XID。
4. 詳細を見てください： チェック SQL_REDO 誤操作発言ですか？確認する USERNAME、SESSION_INFO。
5. 見て SQL_UNDO 空いてますか？：
   • NULLが多い → チェック SUPPLEMENTAL_LOG_DATA_MIN それがYESかどうか。主キー補足ログが有効かどうか。
6. ロールバック スクリプトの生成とレビュー: スクリプトが押されました COMMIT_SCN DESC, RS_ID DESC, SSN DESC 同じトランザクション内で「最後に実行されたものが最初にロールバックされる」ようにするために、逆の順序に配置します。

4.3 ロールバック実行の原則

• まずはバックアップ：CREATE TABLE T_BAK AS SELECT * FROM T;
• 最初は小さなバッチで検証する: 1 ～ 2 項目を選択してください SQL_UNDO 手動で実行し、WHERE 条件がターゲット行にヒットすることを確認します。
• 自動送信をオフにする：SET AUTOCOMMIT OFF;行数を確認してから COMMIT。
• 固有の制約/トリガーに注意する注: INSERT をロールバックするとビジネス トリガーがトリガーされる可能性があるため、必要に応じて最初に無効にします。
• DDL なし SQL_UNDO: 削除/切り捨てられたテーブルでは、フラッシュバック ドロップ、RMAN のポイントインタイム リカバリ、または手動の再構築を使用する必要があります。

4.4 補足分析SQL（詳細テーブル上で直接実行可能）

-- 某张表在时间窗内的完整变更时间线
SELECT scn, change_time, xid, username, operation, sql_redo
  FROM LOGMNR_STAGE_xxxxxx
 WHERE seg_owner="SCOTT" AND seg_name="EMP"
 ORDER BY scn, rs_id, ssn;

-- 找出影响行数最多的 10 个事务
SELECT * FROM (
  SELECT xid, username, seg_owner, seg_name, COUNT(*) cnt,
         MIN(change_time) first_time, MAX(change_time) last_time
    FROM LOGMNR_STAGE_xxxxxx
   GROUP BY xid, username, seg_owner, seg_name
   ORDER BY cnt DESC
) WHERE ROWNUM <= 10;

-- 导出某事务的回滚语句（手工指定 XID）
SELECT sql_undo FROM LOGMNR_STAGE_xxxxxx
 WHERE xid = '01001B00...' AND sql_undo IS NOT NULL
 ORDER BY commit_scn DESC, rs_id DESC, ssn DESC;

5. よくあるエラーと回避方法

実稼働環境のヒント: LogMiner は CPU/IO を集中的に使用する操作であるため、ビジネスの低いピーク時に実行することをお勧めします。可能であれば、Data Guardスタンバイ・データベースまたは独立した分析データベースで実行してください(スタンバイ・データベースには対応するディクショナリとログが必要です)。アーカイブされたログは削除または上書きされていてはなりません。マイニングする前にこれを確認してください V$ARCHIVED_LOG 対象期間中の録画状況は、 A。