本文将为您介绍每5分钟抽取一次数据,待每天00:00的同步任务抽取完成后,对当天共288次同步任务抽取的所有数据进行计算。

实现思路

  1. 创建一个同步任务为上游,一个SQL任务为下游。
  2. 将同步任务的调度时间设置为每5分钟调度一次(开始时间00:00,结束时间23:59,时间间隔5分钟)。
  3. 配置依赖上一周期-本节点,以形成自依赖。
  4. 将SQL任务设置为每天00:00调度一次。

实现原理

在DataWorks调度系统中,下游对上游的依赖遵循原则为:下游任务生成的实例会找到当天离自己最近结束的一个上游实例作为上游依赖,如果上游依赖实例运行成功,才会触发本节点实例运行。如果上游节点每天生成多个实例,则下游无法识别是哪一个实例离它最近结束,导致必须等上游当天生成的所有实例运行完成后才会运行。因此,上游必须配置自依赖,SQL任务在00:00的实例才会准确依赖00:00生成的同步任务实例结束后再运行。

前提条件

开始本实验前,需要首先准备好下述内容。

  • 请确保已拥有阿里云云账号并进行实名认证。详情请参见准备阿里云账号
  • 创建数据库,并准备好源端数据和目标端表。

步骤一:创建业务流程

  1. 登录DataWorks控制台,单击相应工作空间后的进入数据开发
  2. 选择新建 > 业务流程
    业务流程
  3. 填写业务名称描述,单击新建
  4. 新建节点并配置依赖关系。
    配置依赖关系
    说明 节点1用来将每5分钟调度一次的MySQL数据同步至MaxCompute。

    节点2用来汇总MaxCompute接收的数据。

步骤二:配置分钟调度任务

  1. 配置数据来源和数据去向。
    上游为数据同步任务,将MySQL的数据同步至MaxCompute。根据过滤条件过滤每5分钟更新的数据,目标端的分区根据定时时间的前5分钟创建,保证所有数据都写到同一天的分区中。
    分钟任务

    过滤条件为insert_time>=${startTime} and insert_time<${endTime},在右侧调度配置中,配置参数过滤每间隔5分钟的数据。

  2. 字段映射。
    源端和目标端都创建了id、name、insert_time三列字段,insert_time为时间列,数据可以根据时间来过滤。
    字段映射
  3. 单击右侧的调度配置,进入调度配置页面进行设置。
    为过滤条件中的参数赋值:startTime=[yyyymmddhh24miss−5/24/60] endTime=[yyyymmddhh24miss],每个参数间用空格分隔。
    参数赋值
    开始时间是从00:00点开始,间隔5分钟调度一次,为保证一天的实例都运行完,需要设置自依赖。
    自依赖
    说明
    • 上述配置可以保证您一天产生的实例能够依次运行完成,并保存至MaxCompute表同一天的分区中。
    • 如果您的分钟任务中有一个调度任务出错,则设置自依赖后面的实例都不会运行,需要您手动进行处理。
    • 为避免出现上述问题,您可以设置数据过滤为insert_time<${endTime},每次都进行全量同步,只要有成功的便可将endTime数据进行同步,您不需设置自依赖,但会增加数据库的负担。

步骤三:配置天调度任务

下游是一个SQL节点天任务,将workshop_odps_mi一天分区中的数据都过滤出来插入到workshop_odps_dd表中。
天调度任务
insert overwrite table workshop_odps_dd partition (ds=${yestoday})
select id, name,insert_time from workshop_odps_mi where ${startTime}<=ds and ds<${endTime};

调度配置中的赋值为startTime=$[yyyymmddhh24miss-1] endTime=$[yyyymmddhh24miss] yestoday=$[yyyymmdd-1]

将workshop_odps_mi一天分区中的数据都过滤出来插入到workshop_odps_dd表中。如下所示:
insert overwrite table workshop_odps_dd partition (ds=20190320)
select id, name,insert_time from workshop_odps_mi where 20190320000000<=ds and ds<20190321000000;

因为天运行定时时间在分钟任务运行结束后,所以插入workshop_odps_dd分区ds时间要和分钟任务时间在同一天,您将ds时间减1即可。

单击右侧的调度配置,进入调度配置页面。
调度配置

步骤四:查看运行结果

同步任务配置完成后,单击运行,即可查看运行结果。
插入数据