如何使用Airflow在python中每天运行一次函数和其他函数？答案

【问题标题】：How to run a function once in life and other function on daily basis in python using Airflow?如何使用Airflow在python中每天运行一次函数和其他函数？
【发布时间】：2021-03-21 05:42:21
【问题描述】：

我的 python 脚本中有两个函数 1:get_all_invoices() 这个函数从条带下载所有发票，第二个函数:get_daily_invoices() 只下载今天的发票所以我想只运行一次函数 1:get_all_invoices() DAG life 和 2nd-function:get_daily_invoices() 以每日为基础。无论我必须在 python 脚本中管理它还是需要在 Airflow DAG 中管理，可能的解决方案是什么。我的代码：

dag = DAG(
    dag_id=DAG_ID,
    schedule_interval="@daily",
    start_date=START_DATE,
    max_active_runs=MAX_ACTIVE_RUN,
    catchup=False,
    default_args=default_args,
    concurrency=MAX_CONCURRENT_TASKS

)
latest_only = LatestOnlyOperator(task_id='my_latest_only', dag=dag)
stripetos3_scheduler = DockerOperator(
    task_id='stripe-to-s3',
    image='registry.gitlab.com/mobinalhassan/stripetos3dags:latest',
    auto_remove=True,
    force_pull=True,
    dag=dag
)
latest_only >> stripetos3_scheduler

【问题讨论】：

您的代码中的get_daily_invoices() 和get_all_invoices() 在哪里？
据我所知，您有两种选择。
@MenyIssakov 是的，请告诉我...

标签： python python-3.x airflow airflow-scheduler

【解决方案1】：

我不完全明白您是想完全运行第一个函数一次，还是以后可能要运行它。

第一种选择是将这两个函数分成 2 个不同的 dag，一个带有每日时间表，一个没有时间表（手动或通过 API 触发）。

另一种选择是为内部状态（数据库/文件，任何地方..）保存一个标志，这样如果标志已经设置，第一个函数将跳过执行

【讨论】：