Параллельная обработка с помощью Oracle Scheduler

Можно получить выигрыш в производительности, если выполнять обработку данных параллельно несколькими потоками - в нескольких сеансах Oracle. При помощи пакета DBMS_JOB этого можно было добиться и в более ранних версиях Oracle. Однако Oracle Scheduler, появившийся в версии 10 и усовершенствованный в версии 11, предоставляет для этого более развитые возможности. В этой статье я

1. кратко рассмотрю цепочки (chains) Oracle Scheduler,
2. продемонстрирую создание цепочки для параллельной обработки данных,
3. продемонстрирую организацию параллельной обработки данных без создания цепочки.

Цепочка Oracle Scheduler

• позволяет скоординировать выполнение нескольких программ,
• имеет имя,
• создается с помощью dbms_scheduler.create_chain,
• запускается на выполнение с помощью dbms_scheduler.run_chain,
• удаляется с помощью dbms_scheduler.drop_chain.

Цепочка содержит шаги (chain steps) и правила (chain rules).

Шаг цепочки

• определяет, какую программу (или цепочку) необходимо запустить,
• имеет имя,
• создается с помощью dbms_scheduler.define_chain_step,
• запускается на выполнение согласно правилам цепочки,
• удаляется с помощью dbms_scheduler.drop_chain_step.

Правило цепочки

• содержит условие и действие,
• однократно при выполнении условия выполняет действие,
• имеет имя,
• создается с помощью dbms_scheduler.define_chain_rule,
• создается с помощью dbms_scheduler.drop_chain_rule.

Обычно, условия правил проверяют завершение выполнения шагов цепочки, а действия правил - запускают на выполнение шаги цепочки. В каждой правильно построенной цепочке должны быть

• правило с условием, которое выполняется безусловно(!), например, 1=1; это правило запускает на выполнение первый логический шаг цепочки;
• правило с действием END, которое завершает выполнение цепочки.

Прежде чем построить и выполнить демонстрационную цепочку, нужно познакомиться еще с одним объектом Oracle Scheduler - программой (program). Программа

• определяет PL/SQL блок, имя PL/SQL процедуры или внешнюю команду ОС, которую необходимо выполнить,
• имеет имя,
• создается с помощью dbms_scheduler.create_program,
• может иметь параметры, создаваемые с помощью dbms_scheduler.define_program_argument,
• запускается на выполнение из шага цепочки или из задания (scheduler job),
• удаляется с помощью dbms_scheduler.drop_program.

В отличие от шагов и правил, программа непосредственно не связана с конкретной цепочкой, но связывается с шагом цепочки при его определении. Одна и та же программа, при необходимости, может использоваться многими шагами разных цепочек.

Спроектирую цепочку для параллельной обработки данных. Первый шаг цепочки будет выполнять подготовительную работу. Далее будут параллельно запускаться 3 шага, каждый из которых делает свою часть работы. Последний шаг будет выполнять некоторую консолидирующую работу, пользуясь результатами завершившихся параллельных шагов. Например, он может выгрузить подготовленные данные во внешний файл, или послать по электронной почте уведомление о завершении обработки данных.

Для демонстрационных целей создам таблицу demo_data:

SQL> CREATE TABLE demo_data (data VARCHAR2(50));
Table created

Прежде всего, определю программы, которые будут использоваться цепочкой:

SQL> BEGIN
  2   -- подготовка почвы для других программ
  3   dbms_scheduler.create_program(
  4    program_name => 'DEMO_P_START',
  5    program_type => 'PLSQL_BLOCK',
  6    program_action => 'BEGIN DELETE FROM demo_data; END;',
  7    enabled => TRUE
  8   );
  9   -- эти программы параллельно заполняют таблицу demo_data результатами работы
 10   dbms_scheduler.create_program(
 11    program_name => 'DEMO_P_0',
 12    program_type => 'PLSQL_BLOCK',
 13    program_action =>
 14     'INSERT INTO demo_data SELECT ''0''||LEVEL FROM dual CONNECT BY LEVEL <= 5;',
 15    enabled => TRUE
 16   );
 17   dbms_scheduler.create_program(
 18    program_name => 'DEMO_P_1',
 19    program_type => 'PLSQL_BLOCK',
 20    program_action =>
 21     'INSERT INTO demo_data SELECT ''1''||LEVEL FROM dual CONNECT BY LEVEL <= 5;',
 22    enabled => TRUE
 23   );
 24   dbms_scheduler.create_program(
 25    program_name => 'DEMO_P_2',
 26    program_type => 'PLSQL_BLOCK',
 27    program_action =>
 28     'INSERT INTO demo_data SELECT ''2''||LEVEL FROM dual CONNECT BY LEVEL <= 5;',
 29    enabled => TRUE
 30   );
 31   -- последний штрих, используя подготовленные данные
 32   dbms_scheduler.create_program(
 33    program_name => 'DEMO_P_FINALLY',
 34    program_type => 'PLSQL_BLOCK',
 35    program_action => 'UPDATE demo_data SET data = ''#''||data;',
 36    enabled => TRUE
 37   );
 38  END;
 39  /
PL/SQL procedure successfully completed

Для программы типа PLSQL_BLOCK в качестве выполняемого действия можно указать PL/SQL блок, как для DEMO_P_START выше, или фрагмент кода на PL/SQL, не заключенный в ключевые слова BEGIN и END. Перед запуском на выполнение Oracle Scheduler поместит код программы внутрь PL/SQL блока.

Посмотреть созданные программы можно следующим запросом:

SQL> SELECT program_name, program_type, program_action
  2  FROM dba_scheduler_programs
  3  WHERE program_name LIKE 'DEMO_P_%';

PROGRAM_NAME         PROGRAM_TYPE     PROGRAM_ACTION
-------------------- ---------------- --------------------------------------------------------------------------------
DEMO_P_0             PLSQL_BLOCK      INSERT INTO demo_data SELECT '0'||LEVEL FROM dual CONNECT BY LEVEL <= 5;
DEMO_P_1             PLSQL_BLOCK      INSERT INTO demo_data SELECT '1'||LEVEL FROM dual CONNECT BY LEVEL <= 5;
DEMO_P_2             PLSQL_BLOCK      INSERT INTO demo_data SELECT '2'||LEVEL FROM dual CONNECT BY LEVEL <= 5;
DEMO_P_FINALLY       PLSQL_BLOCK      UPDATE demo_data SET data = '#'||data;
DEMO_P_START         PLSQL_BLOCK      BEGIN DELETE FROM demo_data; END;

Создам цепочку DEMO_CHAIN:

SQL> BEGIN
  2      dbms_scheduler.create_chain(
  3          chain_name => 'DEMO_CHAIN'
  4      );
  5  END;
  6  /
PL/SQL procedure successfully completed

Каждый шаг цепочки будет связан с одной из программ, определенных выше. В связи с этим, можно автоматизировать создание шагов цепочки:

SQL> BEGIN
  2      FOR x IN (
  3          SELECT program_name
  4          FROM dba_scheduler_programs
  5          WHERE owner = USER
  6              AND program_name LIKE 'DEMO_P_%')
  7      LOOP
  8          dbms_scheduler.define_chain_step(
  9              chain_name => 'DEMO_CHAIN',
 10              step_name => 'DEMO_S_' || REGEXP_SUBSTR(x.program_name, '[^_]+$'),
 11              program_name => x.program_name
 12          );
 13      END LOOP;
 14  END;
 15  /
PL/SQL procedure successfully completed

В результате определены следующие шаги:

SQL> SELECT step_name, step_type, program_name
  2  FROM dba_scheduler_chain_steps
  3  WHERE chain_name='DEMO_CHAIN';

STEP_NAME            STEP_TYPE      PROGRAM_NAME
-------------------- -------------- --------------------
DEMO_S_0             PROGRAM        DEMO_P_0
DEMO_S_START         PROGRAM        DEMO_P_START
DEMO_S_FINALLY       PROGRAM        DEMO_P_FINALLY
DEMO_S_2             PROGRAM        DEMO_P_2
DEMO_S_1             PROGRAM        DEMO_P_1

Для завершения определения цепочки нужно создать правила. Первое правило запускает на выполнение шаг DEMO_S_START:

SQL> BEGIN
  2      dbms_scheduler.define_chain_rule(
  3          chain_name => 'DEMO_CHAIN',
  4          condition => '1=1',
  5          action => 'START DEMO_S_START',
  6          rule_name => 'DEMO_R_START'
  7      );
  8  END;
  9  /
PL/SQL procedure successfully completed

Следующее правило проверяет, завершен ли шаг DEMO_S_START, и запускает шаги DEMO_S_0, DEMO_S_1 и DEMO_S_2 параллельно:

SQL> BEGIN
  2      dbms_scheduler.define_chain_rule(
  3          chain_name => 'DEMO_CHAIN',
  4          condition => 'DEMO_S_START SUCCEEDED',
  5          action => 'START DEMO_S_0, DEMO_S_1, DEMO_S_2',
  6          rule_name => 'DEMO_R_CONTINUE'
  7      );
  8  END;
  9  /
PL/SQL procedure successfully completed

Третье правило проверяет, завершены ли шаги DEMO_S_0, DEMO_S_1 и DEMO_S_2, и запускает шаг DEMO_S_FINALLY:

SQL> BEGIN
  2      dbms_scheduler.define_chain_rule(
  3          chain_name => 'DEMO_CHAIN',
  4          condition => 'DEMO_S_0 COMPLETED AND DEMO_S_1 COMPLETED AND DEMO_S_2 COMPLETED',
  5          action => 'START DEMO_S_FINALLY',
  6          rule_name => 'DEMO_R_FINALLY'
  7      );
  8  END;
  9  /
PL/SQL procedure successfully completed

Четвертое, и последнее, правило завершает выполнение цепочки после завершения шага DEMO_S_FINALLY:

SQL> BEGIN
  2      dbms_scheduler.define_chain_rule(
  3          chain_name => 'DEMO_CHAIN',
  4          condition => 'DEMO_S_FINALLY COMPLETED',
  5          action => 'END',
  6          rule_name => 'DEMO_R_END'
  7      );
  8  END;
  9  /
PL/SQL procedure successfully completed

Посмотреть созданные правила можно следующим запросом:

SQL> select rule_name, condition, action
  2  from dba_scheduler_chain_rules
  3  where chain_name = 'DEMO_CHAIN';
RULE_NAME            CONDITION                                                              ACTION
-------------------- ---------------------------------------------------------------------- --------------------------------------------------------------------------------
DEMO_R_FINALLY       DEMO_S_0 COMPLETED AND DEMO_S_1 COMPLETED AND DEMO_S_2 COMPLETED       START "DEMO_S_FINALLY"
DEMO_R_END           DEMO_S_FINALLY COMPLETED                                               END
DEMO_R_START         1=1                                                                    START "DEMO_S_START"
DEMO_R_CONTINUE      DEMO_S_START SUCCEEDED                                                 START "DEMO_S_0","DEMO_S_1","DEMO_S_2"

Условия правил используют либо специальный синтаксис (scheduler chain condition syntax), либо синтаксис выражений SQL WHERE. В условиях можно проверять не только завершение шагов как таковое (ИМЯШАГА COMPLETED), но и успешное (ИМЯШАГА SUCCEEDED) или ошибочное завершение шага (ИМЯШАГА FAILED), а также анализировать код ошибки, переданный из программы через RAISE_APPLICATION_ERROR (ИМЯШАГА ERROR_CODE). Об условиях и действиях правил см. описание процедуры dbms_scheduler.define_chain_rule в документации по Oracle 11g.

Прежде, чем цепочку пожно будет запустить на выполнение, необходимо ее активировать:

SQL> BEGIN
  2      dbms_scheduler.enable('DEMO_CHAIN');
  3  END;
  4  /
PL/SQL procedure successfully completed

Теперь запущу цепочку и проверю результат ее работы:

SQL> BEGIN
  2      dbms_scheduler.run_chain('DEMO_CHAIN', '');
  3  END;
  4  /
PL/SQL procedure successfully completed

SQL> select * from demo_data;
DATA
--------------------------------------------------------------------------------
#21
#22
#23
#24
#25
#11
#12
#13
#14
#15
#31
#32
#33
#34
#35
15 rows selected

Успешно!

Посмотреть подробности выполнения шагов цепочки можно следующим запросом:

SQL> SELECT job_name, job_subname, status, actual_start_date, run_duration
  2  FROM dba_scheduler_job_run_details
  3  WHERE job_subname LIKE 'DEMO%'
  4      AND log_date > SYSDATE - 1/96
  5  ORDER BY actual_start_date;

JOB_NAME                       JOB_SUBNAME                    STATUS          ACTUAL_START_DATE                                  RUN_DURATION
------------------------------ ------------------------------ --------------- -------------------------------------------------- --------------------------------------
RUN_CHAIN$DEMO_CHA42864        DEMO_S_START                   SUCCEEDED       22-JAN-15 12.16.43.194386 PM ASIA/VLADIVOSTOK      +000 00:00:00
RUN_CHAIN$DEMO_CHA42864        DEMO_S_0                       SUCCEEDED       22-JAN-15 12.16.43.291043 PM ASIA/VLADIVOSTOK      +000 00:00:00
RUN_CHAIN$DEMO_CHA42864        DEMO_S_1                       SUCCEEDED       22-JAN-15 12.16.43.293827 PM ASIA/VLADIVOSTOK      +000 00:00:00
RUN_CHAIN$DEMO_CHA42864        DEMO_S_2                       SUCCEEDED       22-JAN-15 12.16.43.310581 PM ASIA/VLADIVOSTOK      +000 00:00:00
RUN_CHAIN$DEMO_CHA42864        DEMO_S_FINALLY                 SUCCEEDED       22-JAN-15 12.16.43.745810 PM ASIA/VLADIVOSTOK      +000 00:00:00

Все шаги демонстрационной цепочки отработали быстро, поскольку они выполняют очень мало работы. Если бы цепочка выпонялась долго, то следующий запрос помог бы проследить за текущим статусом выполнения ее шагов:

SQL> SELECT chain_name, step_name, state, error_code, completed, start_date, end_date
  2  FROM dba_scheduler_running_chains
  3  WHERE chain_name = 'DEMO_CHAIN'
  4      AND start_date >= SYSDATE - 1/96
  5  ORDER BY start_date;

no rows selected

На этом закончу работу с цепочкой и удалю ее:

SQL> exec dbms_scheduler.disable('DEMO_CHAIN');
PL/SQL procedure successfully completed

SQL> exec dbms_scheduler.drop_chain('DEMO_CHAIN');
PL/SQL procedure successfully completed

Вместе с цепочкой были удалены ее шаги и правила:

SQL> SELECT step_name, step_type, program_name
  2  FROM dba_scheduler_chain_steps
  3  WHERE chain_name='DEMO_CHAIN';

no rows selected

SQL> select rule_name, condition, action
  2  from dba_scheduler_chain_rules
  3  where chain_name = 'DEMO_CHAIN';

no rows selected

Но программы, непосредственно не связанные с цепочкой, остались, и их нужно удалить отдельно:

SQL> SELECT program_name, program_type, program_action
  2  FROM dba_scheduler_programs
  3  WHERE program_name LIKE 'DEMO_P_%';

PROGRAM_NAME         PROGRAM_TYPE     PROGRAM_ACTION
-------------------- ---------------- --------------------------------------------------------------------------------
DEMO_P_0             PLSQL_BLOCK      INSERT INTO demo_data SELECT '0'||LEVEL FROM dual CONNECT BY LEVEL <= 5;
DEMO_P_1             PLSQL_BLOCK      INSERT INTO demo_data SELECT '1'||LEVEL FROM dual CONNECT BY LEVEL <= 5;
DEMO_P_2             PLSQL_BLOCK      INSERT INTO demo_data SELECT '2'||LEVEL FROM dual CONNECT BY LEVEL <= 5;
DEMO_P_FINALLY       PLSQL_BLOCK      UPDATE demo_data SET data = '#'||data;
DEMO_P_START         PLSQL_BLOCK      BEGIN DELETE FROM demo_data; END;

SQL> BEGIN
  2      FOR x IN (
  3          SELECT * FROM dba_scheduler_programs WHERE program_name LIKE 'DEMO_P_%')
  4      LOOP
  5          dbms_scheduler.drop_program(x.program_name);
  6      END LOOP;
  7  END;
  8  /
PL/SQL procedure successfully completed

Теперь продемонстрирую выполнение такого же сценария - подготовка, параллельная обработка и завершение - без создания цепочки.

Для этого создам пакет DEMO_PKG, в который помещу процедуры

• initialize - для подготовки данных (аналог программы DEMO_P_START),
• process - для обработки данных (аналог программ DEMO_P_0, DEMO_P_1, DEMO_P_2),
• finalize - для завершения обработки (аналог программы DEMO_S_FINALLY),
• run - для скоординированного выполнения процедур (аналог цепочки DEMO_CHAIN).

Для параллельного запуска процедуры process с разными аргументами в нескольких сеансах воспользуюсь dbms_scheduler.create_job. Контроль завершения процедур в параллельных сеансах выполняется с помощью пакета DBMS_LOCK (см. Конкурентный доступ к ресурсам и DBMS_LOCK).

CREATE OR REPLACE PACKAGE demo_pkg IS
    PROCEDURE process(p_param PLS_INTEGER);
    PROCEDURE run;
END demo_pkg;
/

CREATE OR REPLACE PACKAGE BODY demo_pkg IS

    LOCK_NAME CONSTANT VARCHAR2(30) := 'demopkglock';
    
    PROCEDURE initialize IS
    BEGIN
        -- приготовиться
        DELETE FROM demo_data;
    END initialize;
    
    PROCEDURE process(p_param PLS_INTEGER) IS
        l_lock VARCHAR2(30);
        l_status NUMBER;
    BEGIN
        -- поднять флаг обработки
        dbms_lock.allocate_unique(LOCK_NAME, l_lock);
        l_status := dbms_lock.request(l_lock, dbms_lock.s_mode);
        -- выполнить обработку
        INSERT INTO demo_data SELECT p_param||LEVEL FROM dual CONNECT BY LEVEL <= 5;
        dbms_lock.sleep(5);
        -- снять флаг обработки
        l_status := dbms_lock.release(l_lock);
    EXCEPTION
    WHEN OTHERS THEN
        -- снять флаг обработки
        l_status := dbms_lock.release(l_lock);
    END process;
    
    PROCEDURE finalize IS
    BEGIN
        -- могло быть что-то полезное
        UPDATE demo_data SET data = '#'||data;
    END finalize;
    
    PROCEDURE run IS
        l_lock VARCHAR2(30);
        l_status NUMBER;
    BEGIN
        -- подготовить данные для параллельной обработки
        initialize;
        -- запустить параллельную обработку в трех сеансах
        FOR i IN 1..3 LOOP
            dbms_scheduler.create_job(
                job_name => 'demopkgprocess' || i,
                job_type => 'PLSQL_BLOCK',
                job_action => 'demo_pkg.process(' || i || ');',
                enabled => TRUE
            );
        END LOOP;
        -- дать время параллельным задачам получить блокировку
        dbms_lock.sleep(1);
        -- и ждать освобождения блокировки всеми задачами
        dbms_lock.allocate_unique(LOCK_NAME, l_lock);
        l_status := dbms_lock.request(l_lock, dbms_lock.x_mode);
        l_status := dbms_lock.release(l_lock);
        -- завершить обработку
        finalize;
    END run;
        
END demo_pkg;
/

Запущу формирование данных и проверю результат:

SQL> exec demo_pkg.run
PL/SQL procedure successfully completed

SQL> select * from demo_data;
DATA
--------------------------------------------------------------------------------
#11
#12
#13
#14
#15
#21
#22
#23
#24
#25
#31
#32
#33
#34
#35
15 rows selected

Что ж, результат соответствует ожиданиям.

Демонстрационный пакет DEMO_PKG не содержит ни одной команды COMMIT или ROLLBACK, в реальном пакете они могут быть уместны.

В заключение, удаляю следы моих демонстраций:

SQL> DROP PACKAGE demo_pkg;
Package dropped

SQL> DROP TABLE demo_data;
Table dropped