更高效准确的数据库内部任务调度实践,阿里云数据库SelectDB 内核 Apache Doris 内置 Job Scheduler 的实现与应用

本文涉及的产品
阿里云百炼推荐规格 ADB PostgreSQL,4核16GB 100GB 1个月
云原生数据仓库AnalyticDB MySQL版,基础版 8ACU 100GB 1个月
简介: Apache Doris 2.1 引入了内置的 Job Scheduler,旨在解决依赖外部调度系统的问题,提供秒级精确的定时任务管理。

2023 年 3 月,在阿里云瑶池数据库峰会上,阿里云与飞轮科技正式达成战略合作协议,双方旨在共同研发名为“阿里云数据库 SelectDB 版”的新一代实时数据仓库,为用户提供在阿里云上的全托管服务。
SelectDB 是飞轮科技基于 Apache Doris 内核打造的聚焦于企业大数据实时分析需求的企业级产品。因此阿里云数据库 SelectDB 版也延续了 Apache Doris 性能优异、架构精简、稳定可靠、生态丰富等核心特性,同时还融入了云服务随需而用的特性,通过云原生存算分离的创新架构,为企业带来分钟级弹性伸缩、高性价比、简单易用、安全稳定的一键式云上实时分析体验。

为了更深度的了解阿里云数据库 SelectDB 版,我们可以全面多角度的了解 Apache Doris 的应用实践和经验。

在数据管理愈加精细化的需求背景下,定时调度在其中扮演着重要的角色。它通常被应用于以下场景:

  • 定期数据更新,如周期性数据导入和 ETL 操作,减少人工干预,提高数据处理的效率和准确性。
  • 结合 Catalog 实现外部数据源数据定期同步,确保多源数据高效、准确的整合到目标系统中,满足复杂的业务分析需求。
  • 定期清理过期/无效数据,释放存储空间,避免过多过期/无效数据对系统性能产生影响。

在 Apache Doris 之前版本中,通常需要依赖于外部调度系统,如通过业务代码定时调度或者引入第三方调度工具、分布式调度平台来满足上述需求。然而,因受限于外部系统自身能力,可能无法满足 Doris 对调度策略及资源管理灵活性的要求。此外,如果外部调度系统出现故障,这不仅会增加业务风险,还需投入额外的运维时间和人力来应对。

引入 Job Scheduler

为解决上述问题,Apache Doris 在 2.1 版本中引入了 Job Scheduler 功能,实现了自主任务调度能力,调度的精准度可达到秒级。该功能的推出不仅保障了数据导入的完整性和一致性,更让用户能够灵活、便捷调整调度策略。同时,因减少了对外部系统的依赖,也降低了系统故障的风险和运维成本,为社区用户带来更加统一、可靠的使用体验。

Doris Job Scheduler 是一种基于预设计划运行的任务管理系统,能够在特定时间点或按照指定时间间隔触发预定义操作,实现任务的自动化执行。Job Scheduler 具备以下特点:

  • 高效调度:Job Scheduler 可以在指定的时间间隔内安排任务和事件,确保数据处理的高效性。采用时间轮算法保证事件能够精准做到秒级触发。
  • 灵活调度:Job Scheduler 提供了多种调度选项,如按 分、小时、天或周的间隔进行调度,同时支持一次性调度以及循环(周期)事件调度,并且周期调度也可以指定开始时间、结束时间。
  • 事件池和高性能处理队列:Job Scheduler 采用 Disruptor 实现高性能的生产消费者模型,最大可能的避免任务执行过载。
  • 调度记录可追溯:Job Scheduler 会存储最新的 Task 执行记录(可配置),通过简单的命令即可查看任务执行记录,确保过程可追溯。
  • 高可用:依托于 Doris 自身的高可用机制,Job Schedule 可以很轻松的做到自恢复、高可用。

(具体实现原理可参考本文“设计与实现”章节介绍)

语法及示例

01 语法说明

一条有效的 Job 语句需包含以下内容:

  • 关键字 CREATE JOB 需加作业名称,它在数据库中标识唯一事件。
  • ON SCHEDULE 子句用于指定 Job 作业的类型、触发时间和频率。
    • AT timestamp 用于一次性事件。它指定 JOB 仅在给定的日期和时间执行一次,AT CURRENT_TIMESTAMP 指定当前日期和时间。因 JOB 一旦创建则会立即运行,也可用于异步任务创建。
    • EVERY:用于周期性作业,可指定作业的执行频率,关键字后需指定时间间隔(周、天、小时、分钟)。
      • Interval:用于指定作业执行频率。1 DAY 表示每天执行一次,1 HOUR表示每小时执行一次,1 MINUTE 表示每分钟执行一次,1 WEEK 表示每周执行一次。
      • 子句EVERY包含可选 STARTS子句。STARTS后面为 timestamp 值,该值用于定义开始重复的时间,CURRENT_TIMESTAMP 用于指定当前日期和时间。JOB 一旦创建则会立即运行。
      • 子句EVERY包含可选 ENDS子句。ENDS 关键字后面为timestamp 值,该值定义 JOB 事件停止运行的时间。
  • DO 子句用于指定 Job 作业触发时所需执行的操作,目前仅支持 Insert 语句。
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 my_job ON SCHEDULE EVERY 1 MINUTE DO INSERT INTO db1.tbl1 SELECT * FROM db2.tbl2;

该语句表示创建一个名为 my_job的作业,每分钟执行一次,执行的操作是将 db2.tbl2 中的数据导入到 db1.tbl1中。

02 举例说明

创建一次性的 Job: 在 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;

创建周期性的 Job,未指定结束时间: 在 22025-01-01 00:00:00 时开始每天执行 1 次,将 db2.tbl2 中数据导入到 db1.tbl1 中。

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);

创建周期性的 Job,指定结束时间: 在 2025-01-01 00:00:00 时开始每天执行 1 次,将 db2.tbl2 中的数据导入到 db1.tbl1 中,在 2026-01-01 00:10:00 时结束。

CREATE JOB my_job ON SCHEDULER 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 create_time >=  days_add(now(),-1);

借助 Job 实现异步执行: 由于 Job 在 Doris 中是以同步任务的形式创建的,但其执行过程却是异步进行的,这一特性使得 Job 非常适合用于实现异步任务,例如常见的 insert into select 任务。

假设需要将db2.tbl2 中的数据导入到 db1.tbl1 中,这里只需要指定 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 配合,高效完成跨数据源的定期数据同步

基于 Catalog 与 Job Scheduler 的数据自动同步.png

以上表为例,用户希望查询符合总消费金额、最后一次访问时间、性别、所在城市这几个数值条件的用户,并将满足条件的用户信息导入到 Doris 中,以便后续的定向推送。

1. 首先,创建一张 Doris 表

CREATE TABLE IF NOT EXISTS user_activity
(
    `user_id` LARGEINT NOT NULL COMMENT "用户id",
    `date` DATE NOT NULL COMMENT "数据灌入日期时间",
    `city` VARCHAR(20) COMMENT "用户所在城市",
    `age` SMALLINT COMMENT "用户年龄",
    `sex` TINYINT COMMENT "用户性别",
    `last_visit_date` DATETIME REPLACE DEFAULT "1970-01-01 00:00:00" COMMENT "用户最后一次访问时间",
    `cost` BIGINT SUM DEFAULT "0" COMMENT "用户总消费",
    `max_dwell_time` INT MAX DEFAULT "0" COMMENT "用户最大停留时间",
    `min_dwell_time` INT MIN DEFAULT "99999" COMMENT "用户最小停留时间"
)
AGGREGATE KEY(`user_id`, `date`, `city`, `age`, `sex`)
DISTRIBUTED BY HASH(`user_id`) BUCKETS 1
PROPERTIES (
"replication_allocation" = "tag.location.default: 1"
);

2. 其次,创建对应 MySQL 库的 Catalog

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

3. 最后,将 MySQL 数据导入到 Doris 中。采用 Catalog + Insert Into 的方式来导入全量数据,由于全量导入操作可能会引发系统服务波动,通常选择在业务闲暇时进行操作。

  • 一次性调度: 如下方代码所示,使用一次性任务来定时触发全量导入任务,触发时间为凌晨 3:00。
CREATE JOB one_time_load_job
ON SCHEDULE 
AT '2024-8-10 03:00:00'
DO
 INSERT INTO user_activity FROM SELECT * FROM activity.user.activity
  • 周期调度: 用户也可以创建一个周期性的调度任务,定期更新最新的数据。
CREATE JOB schedule_load
ON SCHEDULE EVERY 1 DAY
DO
 INSERT INTO user_activity FROM SELECT * FROM activity.user.activity where create_time >=  days_add(now(),-1)

设计与实现

高效的调度通常伴随着大量的资源消耗,高精度的调度更是如此。传统的实现方式是直接使用 Java 内置的定时调度能力——定时调度线程周期访问,或采用一些定时调度的工具类库,但其在精度以及内存占用上存在较大的问题。为更好保障性能的前提下降低资源的占用,我们选择 TimingWheel 算法与 Disruptor 结合,实现秒级别的任务调度。

设计与实现

具体来说,利用 Netty 的 HashedWheelTimer 实现时间轮算法,Job Manager 会周期性(默认十分钟)地将未来事件放入时间轮中调度。为了保证任务高效触发并避免资源过度占用,采用 Disruptor 构建单生产者多消费者模型。时间轮仅负责触发,并不直接执行任务。对于到期需触发的任务时,会将其放入 Diapatch 线程,由其负责将任务分发至相应的执行线程池,对于需立即执行的任务,则直接将其投递至相应的任务执行线程池中。

对于单次执行事件,将在调度完成后删除事件定义;对于周期性事件,时间轮中的系统事件将定期拉取下一个周期的执行任务。这样可以避免大量任务集中在一个 Bucket 中,减少无意义的遍历、提高处理效率。

而对于事务型任务,Job Scheduler 能够通过与事务的强关联以及事务回调机制,确保事务型任务的执行结果与预期一致,从而保证数据的完整性和一致性。

结束语

Doris Job Scheduler 是一款强大且灵活的任务调度工具,是数据处理中必不可少的功能之一。除了在数据湖分析、内部 ETL 等常见场景的应用外,Job Scheduler 对于异步物化视图的实现也起到关键的作用。异步物化视图是一个预先计算并存储的结果集,其数据更新的频率与源表的变动紧密相关。当源表数据更新频繁时,为确保物化视图中数据保持最新状态,就需要对物化视图定期刷新。因此在 2.1 版本中,我们巧妙地利用 JOB 定时调度功能,保障了物化视图与源表数据的一致性,大幅降低了人工干预的成本。

未来,Doris Job Scheduler 还会支持以下特性:

  • 支持通过 UI 界面查看不同时段执行的任务分布情况。
  • 支持 JOB 流程编排,即 DAG JOB。这意味着我们可以在内部实现数仓任务编排,与 Catalog 功能叠加将会更高效地完成数据处理和分析工作。
  • 支持对导入任务、UPDATE、DELETE 操作进行定时调度。
相关实践学习
如何在云端创建MySQL数据库
开始实验后,系统会自动创建一台自建MySQL的 源数据库 ECS 实例和一台 目标数据库 RDS。
全面了解阿里云能为你做什么
阿里云在全球各地部署高效节能的绿色数据中心,利用清洁计算为万物互联的新世界提供源源不断的能源动力,目前开服的区域包括中国(华北、华东、华南、香港)、新加坡、美国(美东、美西)、欧洲、中东、澳大利亚、日本。目前阿里云的产品涵盖弹性计算、数据库、存储与CDN、分析与搜索、云通信、网络、管理与监控、应用服务、互联网中间件、移动服务、视频服务等。通过本课程,来了解阿里云能够为你的业务带来哪些帮助     相关的阿里云产品:云服务器ECS 云服务器 ECS(Elastic Compute Service)是一种弹性可伸缩的计算服务,助您降低 IT 成本,提升运维效率,使您更专注于核心业务创新。产品详情: https://www.aliyun.com/product/ecs
相关文章
|
18天前
|
消息中间件 数据挖掘 Kafka
Apache Kafka流处理实战:构建实时数据分析应用
【10月更文挑战第24天】在当今这个数据爆炸的时代,能够快速准确地处理实时数据变得尤为重要。无论是金融交易监控、网络行为分析还是物联网设备的数据收集,实时数据处理技术都是不可或缺的一部分。Apache Kafka作为一款高性能的消息队列系统,不仅支持传统的消息传递模式,还提供了强大的流处理能力,能够帮助开发者构建高效、可扩展的实时数据分析应用。
64 5
|
2月前
|
安全 网络协议 应用服务中间件
AJP Connector:深入解析及在Apache HTTP Server中的应用
【9月更文挑战第6天】在Java Web应用开发中,Tomcat作为广泛使用的Servlet容器,经常与Apache HTTP Server结合使用,以提供高效、稳定的Web服务。而AJP Connector(Apache JServ Protocol Connector)作为连接Tomcat和Apache HTTP Server的重要桥梁,扮演着至关重要的角色
74 2
|
13天前
|
存储 SQL Apache
Apache Doris 开源最顶级基于MPP架构的高性能实时分析数据库
Apache Doris 是一个基于 MPP 架构的高性能实时分析数据库,以其极高的速度和易用性著称。它支持高并发点查询和复杂分析场景,适用于报表分析、即席查询、数据仓库和数据湖查询加速等。最新发布的 2.0.2 版本在性能、稳定性和多租户支持方面有显著提升。社区活跃,已广泛应用于电商、广告、用户行为分析等领域。
Apache Doris 开源最顶级基于MPP架构的高性能实时分析数据库
|
7天前
|
关系型数据库 MySQL Linux
Linux环境下MySQL数据库自动定时备份实践
数据库备份是确保数据安全的重要措施。在Linux环境下,实现MySQL数据库的自动定时备份可以通过多种方式完成。本文将介绍如何使用`cron`定时任务和`mysqldump`工具来实现MySQL数据库的每日自动备份。
22 3
|
14天前
|
SQL DataWorks 关系型数据库
阿里云 DataWorks 正式支持 SelectDB & Apache Doris 数据源,实现 MySQL 整库实时同步
阿里云数据库 SelectDB 版是阿里云与飞轮科技联合基于 Apache Doris 内核打造的现代化数据仓库,支持大规模实时数据上的极速查询分析。通过实时、统一、弹性、开放的核心能力,能够为企业提供高性价比、简单易用、安全稳定、低成本的实时大数据分析支持。SelectDB 具备世界领先的实时分析能力,能够实现秒级的数据实时导入与同步,在宽表、复杂多表关联、高并发点查等不同场景下,提供超越一众国际知名的同类产品的优秀性能,多次登顶 ClickBench 全球数据库分析性能排行榜。
|
19天前
|
NoSQL Cloud Native atlas
探索云原生数据库:MongoDB Atlas 的实践与思考
【10月更文挑战第21天】本文探讨了MongoDB Atlas的核心特性、实践应用及对云原生数据库未来的思考。MongoDB Atlas作为MongoDB的云原生版本,提供全球分布式、完全托管、弹性伸缩和安全合规等优势,支持快速部署、数据全球化、自动化运维和灵活定价。文章还讨论了云原生数据库的未来趋势,如架构灵活性、智能化运维和混合云支持,并分享了实施MongoDB Atlas的最佳实践。
|
20天前
|
NoSQL Cloud Native atlas
探索云原生数据库:MongoDB Atlas 的实践与思考
【10月更文挑战第20天】本文探讨了MongoDB Atlas的核心特性、实践应用及对未来云原生数据库的思考。MongoDB Atlas作为云原生数据库服务,具备全球分布、完全托管、弹性伸缩和安全合规等优势,支持快速部署、数据全球化、自动化运维和灵活定价。文章还讨论了实施MongoDB Atlas的最佳实践和职业心得,展望了云原生数据库的发展趋势。
|
1月前
|
存储 分布式计算 druid
大数据-149 Apache Druid 基本介绍 技术特点 应用场景
大数据-149 Apache Druid 基本介绍 技术特点 应用场景
57 1
大数据-149 Apache Druid 基本介绍 技术特点 应用场景
|
23天前
|
SQL Java 数据库
Spring Boot与Flyway:数据库版本控制的自动化实践
【10月更文挑战第19天】 在软件开发中,数据库的版本控制是一个至关重要的环节,它确保了数据库结构的一致性和项目的顺利迭代。Spring Boot结合Flyway提供了一种自动化的数据库版本控制解决方案,极大地简化了数据库迁移管理。本文将详细介绍如何使用Spring Boot和Flyway实现数据库版本的自动化控制。
23 2
|
1月前
|
SQL 关系型数据库 MySQL
Go语言项目高效对接SQL数据库:实践技巧与方法
在Go语言项目中,与SQL数据库进行对接是一项基础且重要的任务
54 11

热门文章

最新文章

推荐镜像

更多