分类目录归档：数据管道
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',
SQL 有很多可用于计数和计算的内建函数。常用的函数有聚合函数、日期和时间函数、转换函数、窗口函数、字符串函数等。
| 名称 | 功能 | 备注 |
|---|---|---|
| AVG | 平均值 | |
| COUNT | 非空值的个数 | |
| FIRST | 第一个记录的值 | |
| LAST | 最后一个记录的值 | |
| MAX() | 最大值 | |
| MIN | 最小值 | |
| SUM | 总和 |
| 名称 | 功能 | 备注 |
|---|---|---|
| CURDATE | 当前日期 | |
| CURTIME | 当前时间 | |
| NOW | 当前日期和时间 | |
| UNIX_TIMESTAMP | 返回 UNIX 时间戳 | |
| DATE_ADD | 将两个日期相加 | |
| DATE_FORMA |
在数据抽取过程中, 如何通过目标数据库最大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']
比如数据库中存在一些表, 这些表名后缀是按天命名的, 现在需要自动抽取汇总到同一个表中, 这样就需要用到循环抽取 smartpip中已有datax组件, 但只能进行单一表格抽取, 不能增加逻辑 所以我们需要使用到smartpip的diy组件功能 首先我们需要新一个datax抽取任务,在这个任务中我们传递参数ZYM, 比如:
#datax job1 ZYM之后再新一建一个diy任务, 实现循环抽取:
ZYM = '202001'
def fun_job2():
job = os.path.join(ETL_FILE_PATH , '项目名/job1.sql')
report为什么建设一下传统的数据中台那么贵, 我给大家分析一下:
//获取变量,以下为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 kettle的数据同步, 你会用到ktr和kjb
这样"执行SQL脚本" 会被执行很多次,
但kjb不一样, kjb会等待一个组件完成再执行另外一个, 所以很多人 会使用kjb 嵌入 ktr的解决方案来做数据抽取后执行脚本
但是如果已经有了任务的调度平台, 再用kjb 来 嵌套 ktr 的解决方案 就会显得臃肿, 所以可以考虑直接在ktr完成
在需要阻塞的地方加入 "阻塞数据" 的组件, 将"Pass all rows