【PL/SQL】DBMS_SCHEDULERでジョブ管理を極める｜CHAINで依存制御・イベント駆動・リトライ自動化・運用オペ完全ガイド

-- 1. 祝日を除外した「営業日のみ」の名前付きスケジュール
BEGIN
  DBMS_SCHEDULER.CREATE_SCHEDULE(
    schedule_name => 'BUSINESS_DAY_0900',
    start_date    => SYSTIMESTAMP,
    repeat_interval =>
      'FREQ=DAILY;BYDAY=MON,TUE,WED,THU,FRI;BYHOUR=9;BYMINUTE=0;' ||
      'EXCLUDE=JP_HOLIDAYS',  -- 事前にJP_HOLIDAYSカレンダーを登録しておく
    comments      => '平日9時（日本の祝日を除く）'
  );
END;
/

-- 2. 複数ジョブから同じスケジュールを参照
BEGIN
  DBMS_SCHEDULER.CREATE_JOB(
    job_name     => 'DAILY_SYNC',
    job_type     => 'STORED_PROCEDURE',
    job_action   => 'PKG_SYNC.RUN_DAILY',
    schedule_name => 'BUSINESS_DAY_0900',   -- 共有スケジュールを参照
    enabled      => TRUE
  );
END;
/

-- ステップ1: チェーン本体の作成
BEGIN
  DBMS_SCHEDULER.CREATE_CHAIN(
    chain_name => 'DAILY_ETL_CHAIN',
    comments   => '日次ETL：抽出→変換→ロード／失敗時はリカバリ通知'
  );
END;
/

-- ステップ2: 各ステップを定義（事前にPROGRAMを作成しておく）
BEGIN
  DBMS_SCHEDULER.DEFINE_CHAIN_STEP('DAILY_ETL_CHAIN','STEP_EXTRACT', 'PRG_EXTRACT');
  DBMS_SCHEDULER.DEFINE_CHAIN_STEP('DAILY_ETL_CHAIN','STEP_TRANSFORM','PRG_TRANSFORM');
  DBMS_SCHEDULER.DEFINE_CHAIN_STEP('DAILY_ETL_CHAIN','STEP_LOAD',    'PRG_LOAD');
  DBMS_SCHEDULER.DEFINE_CHAIN_STEP('DAILY_ETL_CHAIN','STEP_RECOVER', 'PRG_NOTIFY_ERROR');
END;
/

-- ステップ3: 遷移ルールの定義
BEGIN
  -- 開始ルール：最初にEXTRACTを実行
  DBMS_SCHEDULER.DEFINE_CHAIN_RULE(
    'DAILY_ETL_CHAIN',
    condition => 'TRUE',
    action    => 'START STEP_EXTRACT'
  );

  -- EXTRACT成功 → TRANSFORM
  DBMS_SCHEDULER.DEFINE_CHAIN_RULE(
    'DAILY_ETL_CHAIN',
    condition => 'STEP_EXTRACT SUCCEEDED',
    action    => 'START STEP_TRANSFORM'
  );

  -- TRANSFORM成功 → LOAD
  DBMS_SCHEDULER.DEFINE_CHAIN_RULE(
    'DAILY_ETL_CHAIN',
    condition => 'STEP_TRANSFORM SUCCEEDED',
    action    => 'START STEP_LOAD'
  );

  -- どこかで失敗 → リカバリ通知
  DBMS_SCHEDULER.DEFINE_CHAIN_RULE(
    'DAILY_ETL_CHAIN',
    condition => 'STEP_EXTRACT FAILED OR STEP_TRANSFORM FAILED OR STEP_LOAD FAILED',
    action    => 'START STEP_RECOVER'
  );

  -- 正常終了条件：LOADまで成功 または RECOVERで通知完了
  DBMS_SCHEDULER.DEFINE_CHAIN_RULE(
    'DAILY_ETL_CHAIN',
    condition => 'STEP_LOAD SUCCEEDED OR STEP_RECOVER COMPLETED',
    action    => 'END'
  );
END;
/

-- ステップ4: チェーンを有効化＆起動ジョブを作成
BEGIN
  DBMS_SCHEDULER.ENABLE('DAILY_ETL_CHAIN');

  DBMS_SCHEDULER.CREATE_JOB(
    job_name     => 'RUN_DAILY_ETL',
    job_type     => 'CHAIN',
    job_action   => 'DAILY_ETL_CHAIN',
    schedule_name => 'BUSINESS_DAY_0900',
    enabled      => TRUE
  );
END;
/

-- 1. イベントキューを作成
BEGIN
  DBMS_AQADM.CREATE_QUEUE_TABLE(
    queue_table => 'FILE_ARRIVAL_QT',
    queue_payload_type => 'SYS.SCHEDULER_FILEWATCHER_RESULT');
  DBMS_AQADM.CREATE_QUEUE(
    queue_name  => 'FILE_ARRIVAL_Q',
    queue_table => 'FILE_ARRIVAL_QT');
  DBMS_AQADM.START_QUEUE('FILE_ARRIVAL_Q');
END;
/

-- 2. イベント発火条件でジョブを作成
BEGIN
  DBMS_SCHEDULER.CREATE_JOB(
    job_name    => 'FILE_ARRIVAL_JOB',
    job_type    => 'STORED_PROCEDURE',
    job_action  => 'PKG_LOAD.RUN_FOR_FILE',
    event_condition => q'[tab.user_data.file_name LIKE 'sales_%.csv']',
    queue_spec  => 'FILE_ARRIVAL_Q',
    enabled     => TRUE,
    comments    => 'sales_*.csv到着で即起動'
  );
END;
/

-- クレデンシャル作成（OSユーザで監視する）
BEGIN
  DBMS_CREDENTIAL.CREATE_CREDENTIAL(
    credential_name => 'ORACLE_OS_CRED',
    username        => 'oracle',
    password        => '<password>'
  );
END;
/

-- FILE_WATCHERの作成
BEGIN
  DBMS_SCHEDULER.CREATE_FILE_WATCHER(
    file_watcher_name => 'INBOX_WATCHER',
    directory_path    => '/data/inbox',
    file_name         => '*.csv',
    credential_name   => 'ORACLE_OS_CRED',
    destination       => NULL,     -- ローカル監視
    enabled           => TRUE
  );
END;
/

-- FILE_WATCHERイベントをソースとするジョブ
BEGIN
  DBMS_SCHEDULER.CREATE_JOB(
    job_name           => 'INBOX_LOAD_JOB',
    program_name       => 'PRG_LOAD_INBOX',       -- 事前に登録したPROGRAM
    event_condition    => q'[tab.user_data.file_name LIKE '%.csv']',
    queue_spec         => 'INBOX_WATCHER, SYS.SCHEDULER_FILEWATCHER_Q',
    auto_drop          => FALSE,
    enabled            => TRUE
  );
END;
/

BEGIN
  -- 既存ジョブにリトライ属性を設定
  DBMS_SCHEDULER.SET_ATTRIBUTE('DAILY_SYNC', 'RESTARTABLE',     TRUE);
  DBMS_SCHEDULER.SET_ATTRIBUTE('DAILY_SYNC', 'MAX_FAILURES',    3);
  DBMS_SCHEDULER.SET_ATTRIBUTE('DAILY_SYNC', 'MAX_RUNS',        NULL);
  DBMS_SCHEDULER.SET_ATTRIBUTE('DAILY_SYNC', 'MAX_RUN_DURATION',
    INTERVAL '30' MINUTE);      -- 30分で強制停止
  DBMS_SCHEDULER.SET_ATTRIBUTE('DAILY_SYNC', 'RAISE_EVENTS',
    DBMS_SCHEDULER.JOB_FAILED + DBMS_SCHEDULER.JOB_BROKEN
    + DBMS_SCHEDULER.JOB_RUN_COMPLETED);  -- 主要イベントで通知可
END;
/

-- BROKENから復旧する（原因調査後）
BEGIN
  DBMS_SCHEDULER.ENABLE('DAILY_SYNC');   -- 失敗カウンタもリセット
END;
/

-- 重要ジョブ用クラス（リソース高優先度）
BEGIN
  DBMS_SCHEDULER.CREATE_JOB_CLASS(
    job_class_name      => 'CRITICAL_JC',
    resource_consumer_group => 'CRITICAL_GROUP',  -- リソースマネージャと連携
    logging_level       => DBMS_SCHEDULER.LOGGING_FULL,
    log_history         => 365,   -- 1年分の履歴を保持
    comments            => '基幹バッチ用：高優先度'
  );

  DBMS_SCHEDULER.CREATE_JOB_CLASS(
    job_class_name      => 'LOW_JC',
    logging_level       => DBMS_SCHEDULER.LOGGING_RUNS,
    log_history         => 30,
    comments            => '軽量監視用：低優先度'
  );
END;
/

-- ジョブにクラスを割り当て
BEGIN
  DBMS_SCHEDULER.SET_ATTRIBUTE('DAILY_ETL_CHAIN_JOB', 'JOB_CLASS', 'CRITICAL_JC');
  DBMS_SCHEDULER.SET_ATTRIBUTE('HEARTBEAT_JOB',        'JOB_CLASS', 'LOW_JC');
END;
/

-- 1. SMTPサーバを登録（1回だけ）
BEGIN
  DBMS_SCHEDULER.SET_SCHEDULER_ATTRIBUTE('email_server',  'smtp.example.com:25');
  DBMS_SCHEDULER.SET_SCHEDULER_ATTRIBUTE('email_sender', 'alert@example.com');
END;
/

-- 2. 特定ジョブに通知を追加
BEGIN
  DBMS_SCHEDULER.ADD_JOB_EMAIL_NOTIFICATION(
    job_name   => 'DAILY_ETL_CHAIN_JOB',
    recipients => 'ops@example.com, oncall@example.com',
    sender     => 'alert@example.com',
    subject    => 'ETL通知 [{jobname}] - {event_type}',
    body       => 'Job: {jobname}%N状態: {event_type}%Nエラー: {error_message}',
    events     => 'JOB_FAILED, JOB_BROKEN, JOB_OVER_MAX_DUR'
  );
END;
/

-- 通知を削除する場合
-- DBMS_SCHEDULER.REMOVE_JOB_EMAIL_NOTIFICATION('DAILY_ETL_CHAIN_JOB');