气流 ETL 管道 - 在函数中使用计划日期？答案

【问题标题】：Airflow ETL pipeline - using schedule date in functions?气流 ETL 管道 - 在函数中使用计划日期？
【发布时间】：2016-11-22 15:43:40
【问题描述】：

是否可以在 Python 函数中引用 default_args start_date？

default_args = {
    'owner': 'airflow', 
    'depends_on_past': False, 
    'start_date': datetime(2016, 11, 21),
    'email': ['mmm.mm@mmm.com'], 
    'email_on_failure': True, 
    'email_on_retry': True,
    'retries': 1, 
    'retry_delay': timedelta(minutes=1)
}

我的python脚本主要使用子进程来发出这个语句：

query = '"SELECT * FROM {}.dbo.{} WHERE row_date = \'{}\'"'.format(database,                                                             select_database(database)[table_int],
                                                                       query_date)
command = 'BCP {} queryout \"{}\" -t, -c -a 10240 -S "server" -T'.format(query, os.path.join(path, filename))

我要运行的任务是使用 BCP 查询“Select * from table where date = {}”。目前，我的 python 脚本具有日期变量的所有逻辑（默认为昨天）。但是，最好参考 default_arg 并让气流处理日期。

因此，为简化起见，我想使用 default_arg start_date 和 schedule（每天运行）来填写 BCP 命令中的变量。这是正确的方法还是我应该在 python 脚本中保留日期逻辑？

【问题讨论】：

标签： python airflow

【解决方案1】：

这是正确的方法，但您真正需要的是execution_date，而不是start_date。您可以使用provide_context=True 参数通过PythonOperator 中的上下文获取execution_date 作为'ds' 默认变量。 provide_context=True 参数通过 kwargs 参数传递一组在 Jinja 模板中使用的默认变量。您可以在文档的相关部分阅读更多关于默认变量和 Jinja 模板的信息。 https://airflow.incubator.apache.org/code.html#default-variables https://airflow.incubator.apache.org/concepts.html#jinja-templating

您的代码应如下所示：

def query_db(**kwargs):
    #get execution date in format YYYY-MM-DD
    query_date = kwargs.get('ds') 

    #rest of your logic

t_query_db = PythonOperator( task_id='query_db', python_callable=query_db, provide_context=True, dag=dag)

【讨论】：

酷，让我玩这个真正的快速。一开始我很困惑，但后来我在 API 文档中找到了显示“ds”和“yesterday_ds”等的宏部分。
是的，我应该说得更清楚，因为它确实是一个非常令人困惑的主题。我已经更新了答案，所以希望它能更好地解释它。