Последнее обновление 18 мая 2026 г.

Планировщик заданий (Job Scheduler)

Обзор

Встроенный в Apache Doris Job Scheduler — это система управления заданиями, которая выполняет их по заранее заданному расписанию. Она автоматически запускает операции SQL в указанные моменты времени или через фиксированные интервалы без использования внешних инструментов планирования. Начиная с версии 2.1, Job Scheduler обеспечивает точность планирования до секунды.

Типичные сценарии использования:

Периодическая загрузка данных и обработка ETL с меньшим объёмом ручной работы
Периодическая синхронизация данных между несколькими источниками совместно с Multi-Catalog
Периодическое удаление устаревших или некорректных данных для освобождения места в хранилище
Периодическое обновление асинхронных материализованных представлений

Основные возможности:

Возможность	Описание
Точность до секунды	Использует алгоритм TimingWheel для запуска событий с точностью до секунды
Гибкое планирование	Поддерживает однократное и периодическое выполнение; для периодических заданий можно указать время начала и окончания
Высокопроизводительная очередь	Использует Disruptor для построения высокопроизводительной модели с производителями и потребителями, предотвращая перегрузку при выполнении
История выполнения	Хранит записи о последних выполнениях задач Task (количество записей настраивается), которые можно просмотреть с помощью команд
Высокая доступность	Использует собственный механизм высокой доступности Doris и поддерживает автоматическое восстановление после сбоев

Связанная документация: CREATE JOB

Синтаксис

Полный оператор создания задания Job состоит из следующих трёх частей:

CREATE JOB job_name
    ON SCHEDULE schedule
    [COMMENT 'string']
    DO execute_sql;

schedule: {
    AT timestamp
    | EVERY interval
      [STARTS timestamp]
      [ENDS timestamp]
}

interval:
    quantity { WEEK | DAY | HOUR | MINUTE }

Элементы синтаксиса:

Конструкция	Описание
`CREATE JOB job_name`	Задаёт имя задания Job, которое однозначно идентифицирует его в базе данных
`ON SCHEDULE AT timestamp`	Однократное выполнение: задание выполняется один раз в указанный момент времени. Для немедленного выполнения используйте `CURRENT_TIMESTAMP`
`ON SCHEDULE EVERY interval`	Периодическое выполнение: задание повторяется через указанный интервал
`STARTS timestamp`	(Необязательно) Время начала периодического выполнения. Для немедленного запуска используйте `CURRENT_TIMESTAMP`
`ENDS timestamp`	(Необязательно) Время окончания периодического выполнения
`DO execute_sql`	Оператор SQL, выполняемый при срабатывании расписания (в настоящее время поддерживаются только операторы INSERT)

Поддерживаемые единицы интервала: WEEK, DAY, HOUR, MINUTE

Примеры

Однократное задание

Выполнить задание один раз в 2025-01-01 00:00:00, чтобы загрузить данные из db2.tbl2 в db1.tbl1:

CREATE JOB my_job
    ON SCHEDULE AT '2025-01-01 00:00:00'
    DO INSERT INTO db1.tbl1 SELECT * FROM db2.tbl2;

Периодическое задание (без времени окончания)

Начиная с 2025-01-01 00:00:00, выполнять инкрементальную загрузку один раз в день:

CREATE JOB my_job
    ON SCHEDULE EVERY 1 DAY
    STARTS '2025-01-01 00:00:00'
    DO INSERT INTO db1.tbl1
       SELECT * FROM db2.tbl2
       WHERE create_time >= days_add(now(), -1);

Периодическое задание (со временем окончания)

Выполнять ежедневную загрузку начиная с 2025-01-01 и автоматически прекратить её в 2026-01-01 00:10:00:

CREATE JOB my_job
    ON SCHEDULE EVERY 1 DAY
    STARTS '2025-01-01 00:00:00'
    ENDS '2026-01-01 00:10:00'
    DO INSERT INTO db1.tbl1
       SELECT * FROM db2.tbl2
       WHERE create_time >= days_add(now(), -1);

Асинхронное выполнение

Задания Job создаются в Doris синхронно, а само выполнение происходит асинхронно. Поэтому Job подходят для асинхронных задач, таких как длительные операции INSERT INTO SELECT.

Задайте время начала с помощью CURRENT_TIMESTAMP, и задание Job запустится асинхронно сразу после создания:

CREATE JOB my_job
    ON SCHEDULE AT CURRENT_TIMESTAMP
    DO INSERT INTO db1.tbl1 SELECT * FROM db2.tbl2;

Автоматическая синхронизация данных с помощью Catalog и Job Scheduler

Рассмотрим сценарий электронной коммерции, в котором нужно периодически извлекать бизнес-данные из MySQL и синхронизировать их с Doris для аналитики, используемой в целевом маркетинге. Job Scheduler совместно с Multi-Catalog позволяет эффективно выполнять периодическую синхронизацию данных между источниками.

Шаг 1: Подготовьте исходные данные в MySQL

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

CREATE TABLE IF NOT EXISTS user.activity (
    `user_id`        INT      NOT NULL,
    `date`           DATE     NOT NULL,
    `city`           VARCHAR(20),
    `age`            SMALLINT,
    `sex`            TINYINT,
    `last_visit_date` DATETIME DEFAULT '1970-01-01 00:00:00',
    `cost`           BIGINT   DEFAULT '0',
    `max_dwell_time` INT      DEFAULT '0',
    `min_dwell_time` INT      DEFAULT '99999'
);

INSERT INTO user.activity VALUES
    (10000, '2017-10-01', 'Beijing',   20, 0, '2017-10-01 06:00:00', 20,  10, 10),
    (10000, '2017-10-01', 'Beijing',   20, 0, '2017-10-01 07:00:00', 15,  2,  2),
    (10001, '2017-10-01', 'Beijing',   30, 1, '2017-10-01 17:05:00', 2,   22, 22),
    (10002, '2017-10-02', 'Shanghai',  20, 1, '2017-10-02 12:59:00', 200, 5,  5),
    (10003, '2017-10-02', 'Guangzhou', 32, 0, '2017-10-02 11:20:00', 30,  11, 11),
    (10004, '2017-10-01', 'Shenzhen',  35, 0, '2017-10-01 10:00:00', 100, 3,  3),
    (10004, '2017-10-03', 'Shenzhen',  35, 0, '2017-10-03 10:20:00', 11,  6,  6);

user_id	date	city	age	sex	last_visit_date	cost	max_dwell_time	min_dwell_time
10000	2017-10-01	Beijing	20	0	2017-10-01 06:00:00	20	10	10
10000	2017-10-01	Beijing	20	0	2017-10-01 07:00:00	15	2	2
10001	2017-10-01	Beijing	30	1	2017-10-01 17:05:00	2	22	22
10002	2017-10-02	Shanghai	20	1	2017-10-02 12:59:00	200	5	5
10003	2017-10-02	Guangzhou	32	0	2017-10-02 11:20:00	30	11	11
10004	2017-10-01	Shenzhen	35	0	2017-10-01 10:00:00	100	3	3
10004	2017-10-03	Shenzhen	35	0	2017-10-03 10:20:00	11	6	6

Цель состоит в том, чтобы с помощью запроса выбрать пользователей, соответствующих условиям по сумме расходов, времени посещения, полу и городу, и загрузить их данные в Doris для последующего формирования персонализированных рекомендаций.

Шаг 2: Создайте целевую таблицу в Doris

CREATE TABLE IF NOT EXISTS user_activity (
    `user_id`        LARGEINT NOT NULL COMMENT "User id",
    `date`           DATE     NOT NULL COMMENT "Data ingestion date",
    `city`           VARCHAR(20)      COMMENT "User city",
    `age`            SMALLINT         COMMENT "User age",
    `sex`            TINYINT          COMMENT "User gender",
    `last_visit_date` DATETIME REPLACE DEFAULT "1970-01-01 00:00:00" COMMENT "Last visit time",
    `cost`           BIGINT SUM DEFAULT "0"     COMMENT "Total spending",
    `max_dwell_time` INT    MAX DEFAULT "0"     COMMENT "Maximum dwell time",
    `min_dwell_time` INT    MIN DEFAULT "99999" COMMENT "Minimum dwell time"
)
AGGREGATE KEY(`user_id`, `date`, `city`, `age`, `sex`)
DISTRIBUTED BY HASH(`user_id`) BUCKETS 1
PROPERTIES (
    "replication_allocation" = "tag.location.default: 1"
);

Шаг 3: Создайте Catalog для MySQL

CREATE CATALOG activity PROPERTIES (
    "type"       = "jdbc",
    "user"       = "root",
    "password"   = "123456",
    "jdbc_url"   = "jdbc:mysql://127.0.0.1:3306/user?useSSL=false",
    "driver_url" = "mysql-connector-java-5.1.49.jar",
    "driver_class" = "com.mysql.jdbc.Driver"
);

Шаг 4: Создайте задания Job для синхронизации данных по расписанию

Полная загрузка может вызывать колебания нагрузки на систему, поэтому её обычно планируют на часы низкой нагрузки, например на раннее утро.

Однократное выполнение (полная загрузка, запускаемая один раз в 3:00 утра):

CREATE JOB one_time_load_job
    ON SCHEDULE AT '2024-08-10 03:00:00'
    DO INSERT INTO user_activity
       SELECT * FROM activity.user.activity;

Периодическое выполнение (ежедневная инкрементальная синхронизация последних данных):

CREATE JOB schedule_load
    ON SCHEDULE EVERY 1 DAY
    DO INSERT INTO user_activity
       SELECT * FROM activity.user.activity
       WHERE last_visit_date >= days_add(now(), -1);

Архитектура и реализация

Высокая точность планирования сопряжена с большим потреблением ресурсов. Традиционные решения Java на основе потоков с планированием не обеспечивают достаточной точности и расходуют слишком много памяти. Для решения этой проблемы Job Scheduler сочетает алгоритм TimingWheel и Disruptor, обеспечивая высокую производительность при низком потреблении ресурсов.

Основные механизмы:

Запуск событий с помощью колеса таймеров: алгоритм колеса таймеров реализован с использованием HashedWheelTimer из Netty. Job Manager заранее загружает будущие события в колесо таймеров для планирования каждые десять минут (по умолчанию).
Диспетчеризация через Disruptor: колесо таймеров только инициирует события и не выполняет задачи напрямую. Задачи, для которых наступило время выполнения, сначала поступают в поток Dispatch, который передаёт их в соответствующий пул потоков выполнения. Задачи, требующие немедленного выполнения, направляются непосредственно в пул потоков выполнения.
Управление жизненным циклом событий: определения событий однократных задач автоматически удаляются после завершения планирования. Для периодических задач колесо таймеров регулярно получает план выполнения на следующий цикл. Это позволяет избежать скопления большого количества задач в одной корзине и повышает эффективность обработки.
Транзакционная согласованность: для транзакционных задач Job Scheduler использует тесную связь с транзакциями и механизм обратных вызовов, чтобы результаты выполнения задач соответствовали ожиданиям, а целостность данных сохранялась.

Планы развития

Job Scheduler также играет ключевую роль при работе с асинхронными материализованными представлениями. Если данные исходной таблицы часто меняются, Job Scheduler периодически запускает обновление материализованного представления, чтобы поддерживать его согласованность с источником.

В будущих версиях планируется добавить следующие возможности:

Визуализация распределения задач: пользовательский интерфейс для просмотра распределения выполнений задач по различным временным окнам.
Оркестрация рабочих процессов DAG: поддержка оркестрации рабочих процессов JOB (DAG JOB), позволяющая управлять зависимостями задач хранилища данных внутри Doris. В сочетании с Catalog это повысит эффективность обработки и анализа данных.
Поддержка дополнительных операций: выполнение по расписанию таких операций, как UPDATE и DELETE.

Обзор​

Синтаксис​

Примеры​

Однократное задание​

Периодическое задание (без времени окончания)​

Периодическое задание (со временем окончания)​

Асинхронное выполнение​

Автоматическая синхронизация данных с помощью Catalog и Job Scheduler​

Архитектура и реализация​

Планы развития​