如何用 Apache DolphinScheduler 实现自动配置自定义变量？

海豚调度 2023-08-01

730

点击蓝字关注我们

作者 | sqlboy-yuzhenc

背景介绍

在实际工作应用中，我们可能需要在整个 Apache DolphinScheduler 层面定义共用的日期变量，但是，开源版 Apache DolphinScheduler 只允许配置工作流级别的全局变量，这样，每个工作流就得配置相同的变量，操作起来重复并且麻烦。通过这篇文章，只需要将海豚调度级别的变量插入到变量表中，就会在保存工作流的时候将变量表中的变量自动配置。

原理

在工作流定义表（t_ds_process_definition）上定义一个事前触发器，在更新或插入数据之前，触发器会将t_ds_process_definition.global_params
的值替换为用户在工作流保存界面上配置的全局变量与在变量定义表中定义的变量的并集。

Tool模式操作

创建自定义枚举类型

create type vartype as enum ('INTEGER','VARCHAR','LONG','FLOAT','DOUBLE','DATE','TIME','TIMESTAMP','BOOLEAN','LIST');
create type vardirect as enum ('IN','OUT');

创建变量定义表

drop table if exists prop_def cascade;
create table prop_def (
     prop varchar(63) primary key  --参数名 
    ,prop_def text  --参数定义
    ,prop_def_dp text  --对应的海豚的定义
    ,prop_desc text  --参数描述
    ,prop_direct vardirect default 'IN'  --参数类型  IN或out
    ,prop_type vartype default 'VARCHAR'  --参数类型 VARCHAR INTEGER LONG FLOAT DOUBLE DATE TIME TIMESTAMP BOOLEAN LIST
    ,prop_create_time timestamp default current_timestamp  --变量创建时间
);
comment on table prop_def is '参数定义表';
comment on column prop_def.prop is '参数名';
comment on column prop_def.prop_def is '参数定义';
comment on column prop_def.prop_def_dp is '对应的海豚的参数定义';
comment on column prop_def.prop_direct is '参数类型  IN或out';
comment on column prop_def.prop_type is '参数类型 VARCHAR INTEGER LONG FLOAT DOUBLE DATE TIME TIMESTAMP BOOLEAN LIST';
comment on column prop_def.prop_desc is '参数描述';
comment on column prop_def.prop_create_time is '参数创建时间';

插入变量定义

insert into prop_def(
     prop  --参数名
    ,prop_def  --参数定义
    ,prop_def_dp  --对应的海豚的定义
    ,prop_desc  --参数描述
) values 
 ('today',$$to_char(current_date,'yyyymmdd')$$,$$$[yyyyMMdd]$$,'今天')
,('yesterday',$$to_char(current_date-1,'yyyymmdd')$$,$$$[yyyyMMdd-1]$$,'昨天')
,('tomorrow',$$to_char(current_date+1,'yyyymmdd')$$,$$$[yyyyMMdd+1]$$,'明天')
,('thismonth',$$to_char(current_date,'yyyymm')$$,$$$[yyyyMM]$$,'本月')
,('lastmonth',$$to_char(to_date(to_char(current_date,'yyyymm')||'01','yyyymmdd')-1,'yyyymm')$$,$$$[add_months(yyyyMM,-1)]$$,'上月')
,('nextmonth',$$(to_char(current_date,'yyyymm')::numeric+1)::varchar$$,$$$[add_months(yyyyMM,1)]$$,'下月')
,('thisyear',$$to_char(current_date,'yyyy')$$,$$$[yyyy]$$,'本年')
,('nextyear',$$(to_char(current_date,'yyyy')::int+1)::varchar$$,$$$[add_months(yyyy, 12*1)]$$,'明年')
,('lastyear',$$(to_char(current_date,'yyyy')::int-1)::varchar$$,$$$[add_months(yyyy, -12*1)]$$,'去年')
;

创建获取变量值函数

create or replace function get_propdef(propdef_sql varchar) 
 returns varchar
 language plpgsql
as $function$ 
/*
 * 作者：v-yuzhenc
 * 功能：返回指定变量定义具体实时的值
 * propdef_sql：变量定义sql
 * */
declare 
    p_result varchar;
    p_propdef_sql varchar;
begin
    execute 'select '||propdef_sql into p_result;
       return p_result;
    exception when others then 
        return p_result;
end;
$function$
;
grant execute on function get_propdef(varchar) to public;

创建实时变量值视图

create or replace view v_prop_def as
select 
     current_timestamp as currenttime  --当前时间
    ,prop  --参数名
    ,prop_def  --参数定义
    ,prop_def_dp  --对应的海豚的定义
    ,get_propdef(prop_def) as prop_value  --当前时间的参数值
    ,prop_desc  --参数描述
    ,prop_direct  --参数类型  IN或out
    ,prop_type  --参数类型 VARCHAR INTEGER LONG FLOAT DOUBLE DATE TIME TIMESTAMP BOOLEAN LIST
from prop_def
;
comment on view v_prop_def is '实时变量表';
comment on column v_prop_def.currenttime is '当前时间';
comment on column v_prop_def.prop is '参数名';
comment on column v_prop_def.prop_def is '参数定义';
comment on column v_prop_def.prop_def_dp is '对应的海豚的定义';
comment on column v_prop_def.prop_value is '当前时间的参数值';
comment on column v_prop_def.prop_desc is '参数描述';
comment on column v_prop_def.prop_direct is '参数类型  IN或out';
comment on column v_prop_def.prop_type is '参数类型 VARCHAR INTEGER LONG FLOAT DOUBLE DATE TIME TIMESTAMP BOOLEAN LIST';

dp模式操作

创建触发器函数

create or replace function tg_ds_udef_prop_def()
returns trigger 
as $trigger$
/*
 * 作者：v-yuzhenc
 * 功能：海豚调度自动配置替代变量
 * */
begin
    select (('['||string_agg($${"prop":"$$||prop||$$","value":"$$||value||$$","direct":"$$||direct||$$","type":"$$||type||$$"}$$,',')||']')::json)::text
    into new.global_params
	from (
	    select 
		     coalesce(a.prop,b.prop) prop
		    ,coalesce(a.prop_def_dp,b.value) value
		    ,coalesce(a.prop_direct::varchar,b.direct) direct
		    ,coalesce(a.prop_type::varchar,b.type) type
		from tool.prop_def a 
		full join (
		    select 
		         json_array_elements(new.global_params::json) ->> 'prop' as prop
			    ,json_array_elements(new.global_params::json) ->> 'value' as value
			    ,json_array_elements(new.global_params::json) ->> 'direct' as direct
			    ,json_array_elements(new.global_params::json) ->> 'type' as type
		) b 
		on (a.prop = b.prop)
	) a
	;
    return new;
end;
$trigger$
language plpgsql;

注：这里的tool和dp是pg的模式，我海豚调度的元数据存储在dp模式下，但是变量定义表存储在tool下，所以在函数中访问变量定义表时加了tool.(即tool.prop_def)，请根据自己实际情况修改。

创建触发器

这里创建事前触发器，在更改或者新增一个工作流之前，将变量的定义给到全局参数定义字段

create trigger tg_prop_ds_process_definition before update or insert on t_ds_process_definition for each row execute procedure tg_ds_udef_prop_def();

测试

查询实时变量值视图

select * from v_prop_def;

工作流测试

创建工作流，选择shell插件

echo ${today}
echo ${yesterday}
echo ${tomorrow}
echo ${thismonth}
echo ${lastmonth}
echo ${nextmonth}
echo ${thisyear}
echo ${nextyear}
echo ${lastyear}