如何选择 Apache Airflow 调度程序更新 DAG 的频率？

Question

如Apache Airflow documentation 中所述，我可以通过在airflow.cfg 文件中设置配置变量min_file_process_interval 来控制DAG 的更新频率：

min_file_process_interval

解析 DAG 文件的秒数。 DAG 文件每隔 min_file_process_interval 秒解析一次。对 DAG 的更新会在此时间间隔后反映出来。将此数字保持在较低水平会增加 CPU 使用率。

但是，我没有找到任何线索或最佳实践来说明我应该为 min_file_process_interval 设置哪个值。

示例

我的 DAG 每天更改一次。默认情况下，min_file_process_interval 设置为 30 秒。这意味着大多数时候更新 DAG 是没有用的：只要 DAG 不改变，更新的 DAG 和之前的 DAG 是一样的。它消耗资源并生成日志。但是，如果我每天只更新一次 DAG，如果 DAG 在每日 DAG 更新之后发生更改，或者 DAG 在运行前也更新了，我是否有运行错误 DAG 的风险？

在这种情况下我应该为min_file_process_interval 设置什么值？

编辑：正如Elad's answer 回复此问题的先前版本中所述，应避免使用动态 DAG。但是，如果我有动态 DAG，如何选择min_file_process_interval？

Answer 1

你正在混合两种不同的东西。 min_file_process_interval 表示 Airflow 多久扫描一次 .py 文件并更新 Airflow 中的 DAG。考虑一下，当您部署新的 .py 文件时，Airflow 需要读取它并在元存储数据库中创建它 - 所以设置是关于它发生的频率。

对于您的用例，DAG 代码不应该每天都更新 - 事实上它根本不应该更新。它应该每天运行。您的 dag 只需要能够在每个日期处理正确的文件。你的代码可以是这样的：

from airflow.providers.ftp.sensors.ftp import FTPSensor

with DAG(dag_id='stackoverflow',
         default_args=default_args,
         schedule_interval="@daily",
         catchup=False
         ) as dag:
    # Waits for a file or directory to be present on FTP.
    sensor_op = FTPSensor(
        task_id='sensor_task',
        path='/my_folder/{{ ds }}/file.csv', #path to your file in the server
        fail_on_transient_errors=False,
        ftp_conn_id='ftp_default'
    )
    
    # Operator to process the file
    operator_op = SomeOperator()

    sensor_op >> operator_op

在该 DAG 中，它将每天开始运行 - 第一个操作员是传感器，因此如果当天的文件不存在，工作流将等到它出现时，工作流将继续到第二个操作员，这应该处理文件。 请注意，FTPSensor 的path 参数是模板化的。这意味着您可以像 {{ ds }} 一样使用 macros 这将为您提供一个包含每天日期的路径，例如：

/my_folder/2021-05-01/file.csv
/my_folder/2021-05-02/file.csv
/my_folder/2021-05-03/file.csv

您也可以使用path='/my_folder/{{ ds }}.csv'，这将提供：

/my_folder/2021-05-01.csv
/my_folder/2021-05-02.csv
/my_folder/2021-05-03.csv