-- 当前时间
report_time = datetime.datetime.now()
-- 当前时间,往后推一天, 也可以小时(hours), 分钟(minutes), 秒(seconds)...
report_time = datetime.datetime.now()- datetime.timedelta(days=1)
-- 获取上个月最后一天
report_time = datetime.datetime.now().replace(day=1) - datetime.timedelta(days=1)
-- 格式化字符串时间:(%Y-%m-%d %H:%M:%S):  2015-03-08 23:30:42
ZYMD= report_time.strftime('%Y%m%d')
-- 也可以采用格式化参数f"xxxxxx",变量用{}包裹
PARAM=f"where create_time >= '{ZYMD}' "

-- 每一个JOB为一个节点,填写驱动名称和要执行的任务名
-- 任务名不能使用数字开头如00xxx, 不可以使用非法字符, 不可使用符号如(.)
#starrocks_sql  sql_filename
-- 参数和备注
#impala_sql  sql_filename ZYMD,MSG    -- 我是备注

-- DAG的设定方法
JOB写完在下方输入////, 4个以上/分隔为DAG设定
后可输入如: sql_filename1 >> sql_filename2 >> sql_filename3  为一个分支

使用方法: #dataset jobname id remark[可选] maillist[可选] 
remark参数说明: 
      不填写 - 默认无查询结果抛出错误
      info - 只打印查询结果,
      e1 - 如果有返回结果抛出错误, 
      e2 - 邮件通知查询结果(可以用于需要定时邮件收到数据的需求)
      e3 - 无查询结果时,每15分钟会重试,如2小时还无结果抛出错误,
           可用于轮询业务系统, 等待业务系统数据ready后再抽取

id参数: 可以是数据集ID也可以填数据集的名称, 
maillist: 格式 xxx@xxxx,xxx@xxx

你也可以采用高级自定义，使用
get_dataset(id, param=None):传入数据集ID, 可传参如{"p1":"xxx"}
返回值: {"result":"success","data":[[]]}

使用方法 #link DAG的名字  sleeptime[可选]  maxtime[可选]
    #link DAG_A  --如DAG_A Fail或进行中,此节点后的流程不会再执行
    #link DAG_A  60  3600  --如DAG_A还在进行中, 每60秒检查一次
       如有完成则继续任务, 重试达到3600秒后, 还未完成, 则报错

'''
编写自定义函数名为：任务名 ，基于条件自定义返回的分支进行后续执行，未返回的分支将不会执行
'''
def inv_validate():
    dataset = get_dataset(407)
    if len(dataset['data']) > 1:
        return 'task1'
    return 'task2'

#impala_sql task1
#trigger  task2  targetdagname
#branch  branch_job  inv_validate

/////////////////////////

branch_job  >> [task1, task2]

-- 自定义你的执行函数
def refresh_XX():
    import requests
    response=requests.get('',verify=False).json()
    if response['status']!=200:
        raise Exception(str(response))

#diy refresh_job  refresh_XX  -- 注意: jobname不可与函数名同名

#ktr abc
////////
abc.queue = 'bg'