分类目录归档：smartpip
airflow支持基于时间的任务回跑, 这样我们就需要使用到airflow自带的参数, 如 "execution_date"
airflow能向任务传递的参数如下:
{'conf': , 'dag': , 'dag_run':,
'ds': '2022-07-13',
'ds_nodash': '20220713',
'execution_date': DateTime(2022, 7, 13, 2, 4, 33, 244294, tzinfo=Timezone('+00:00')),
'inlets': [],
'macros': ,
'next_ds': '2022-07-13',
在数据抽取过程中, 如何通过目标数据库最大ID或时间等条件进行抽取, 这样可以将数据同步的粒度做得更细
select
max(id) as maxid, max(event_day) as event_day
from targettablename
#datax jobname 455 -- 同步表sel你可能需要自定义邮件内容, 甚至可能需要动态获取数据进行发送
msg = ['报表刷新成功', '<h1>刷新成功!!</h1>']
maillist = 'xxx@smartchart.cn'
#send_mail reportmail msg maillist
maillist = 'xxx@smartchart.cn,abc@xxx.cn'
def fun_msg():
result = get_dataset(123)['data']
//获取变量,以下为python语法(option)
P_DAYS = 12
MSG = '------------------------------------------------------------'
report_time = datetime.datetime.now()- datetime.timedelta(days=int(P_DAYS))
P_START_ZYM= report_time.strftime('%Y%m')
//以下为JOB专署语法
#link lastdag 30 3600 --填写以下内容到你的SQL文件, 比如命名为: mysqoop.sql
/*
conn = zspl -- 连接串, 找管理员要, 也可自定义, (必填)
sourceTable = tablename -- 源系统表名,(必填)
columns = -- 抽取原表的字段 a,b,c (可省略)
where = -- 抽取时的条件 a>1 (可省略)
seq = -- 分隔符, 默认 \t (可省略)
query = -- 查询语名,Starrocks支持例行导入（Routine Load）功能，提交一个常驻的导入任务，通过不断的从指定的数据源读取数据，将数据导入到 StarRocks 中。
smartpip是我们基于airflow研发的任务调度平台, 针对定时导入kafka数据到starrocks, 已实现了一个配置即可实现 但是如果对实时监听kafka导入, 并不太合适, 所以本文将介绍如何使用starrocks自带的routine load的功能,同时来使用smartpip实现监控功能
在smartchart中新建一个数据集, 查询内容:
SHOW -- 当前时间
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-