气流默认变量-增量负载设置
气流默认变量-增量负载设置
提问于 2020-11-02 17:46:45
我正在尝试为从rds postgres到另一个postgres rds的数据提取实现增量数据加载。
我使用airflow来实现ETL。因此，在阅读了一段时间有关airflow宏的内容后，我决定使用airflow默认变量来设置增量流。
所以，算法是这样的，
如果我的上一次执行日期是None或null或''：从时间的开始(在我们的例子中是一年前)选择数据，否则选择上一次执行日期end，如果
注意:下面的代码首先是为了理解默认变量，这还没有针对我上面提到的问题实现
相应的代码如下所示。当我第一次运行dag时，我总是为之前的成功执行日期变量打印'None‘，而不是像我提到的那样打印历史日期。我无法弄清楚这一点。任何关于这方面的想法都会有很大的帮助
from datetime import *
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.operators.dummy_operator import DummyOperator
from airflow.utils.dates import days_ago
default_args={'owner':'airflow','start_date': days_ago(1),'depends_on_past':'False'}
dag = DAG('jinja_trial_10',default_args=default_args,schedule_interval=timedelta(minutes=5))
def printexecutiontimes(**kwargs):
executiondate = kwargs.get('execution_date')
previoussuccessfulexecutiondate = kwargs.get('prev_execution_date_success')
previousexecutiondate = kwargs.get('prev_ds_nodash')
if (previoussuccessfulexecutiondate == 'None' or previoussuccessfulexecutiondate is None):
previoussuccessfulexecutiondate = datetime.strftime(datetime.now() - timedelta(days = 365),'%Y-%m-%d')
print('Execution Date : {0}'.format(executiondate))
print('Previous successful execution date : {0}'.format(previoussuccessfulexecutiondate))
print('Previous execution date : {0}'.format(previousexecutiondate))
print('hello')
task_start = DummyOperator(task_id = 'start',dag=dag)
jinja_task= PythonOperator(task_id = 'TryingoutJinjatemplates',
python_callable =printexecutiontimes,
provide_context = 'True',
dag=dag )
task_end = DummyOperator(task_id = 'end',dag=dag)
task_start >>jinja_task >> task_endStack Overflow用户
发布于 2020-11-02 19:15:51
我最近不得不做一些非常类似的事情，下面的代码是我最终使用DagRun details创建自定义函数的代码。
Refer to this answer -如果您只想获取上次运行的DAG (无论状态如何)。
对于我来说，我必须获得最后一次成功运行的日期，因此创建了以下函数：
def get_last_dag_run(dag_id):
dag_runs = DagRun.find(dag_id=dag_id)
dag_runs.sort(key=lambda x: x.execution_date, reverse=True)
for dag_run in dag_runs:
#print all dag runs - debug only
print(f"All ----- state: {dag_run.state} , run_id: {dag_run.run_id} , execution_date: {dag_run.execution_date}")
print('Success runs ---------------------------------')
dag_runs = list(filter(lambda x: x.state == 'success', dag_runs))
for dag_run in dag_runs:
#print successfull dag runs - debug only
print(f"Success - state: {dag_run.state} , run_id: {dag_run.run_id} , execution_date: {dag_run.execution_date}")
# return last execution run or default value (01-01-1970)
return dag_runs[0].execution_date if dag_runs else datetime(1970, 1, 1)页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/64643140
复制相关文章
相似问题
腾讯云开发者
