Airflow 2.x 中 DAG 导入错误的日志消息

我正在本地运行 Apache Airflow 2.x，使用中提供的 Docker Compose 文件文档 https://airflow.apache.org/docs/apache-airflow/stable/start/docker.html。在里面.\dags在我的本地文件系统（安装到 Airflow 容器中）的目录中，我创建一个新的 Python 脚本文件，并使用 TaskFlow API 实现 DAG。

我的 DAG 的更改有时无效。例如，也许我有一个ImportError由于模块名称无效或语法错误。当 Airflow 尝试导入 DAG 时，我无法从 Web 服务器、调度程序或工作线程中找到任何表明存在问题或具体问题是什么的日志消息。

相反，我必须逐行阅读代码，并查找问题。由于我在 Windows 10 上的本地 Python 环境和 Airflow 的 Python 环境是不同的版本并且安装了不同的 Python 包，因此这个问题变得更加复杂。因此，我无法可靠地使用本地开发环境来检测包导入失败，因为我期望在 Airflow 环境中安装的包与我本地的包不同。此外，我用于本地编写代码的 Python 版本与 Airflow 使用的 Python 版本不匹配。

因此，我需要某种错误日志记录来指示 DAG 导入失败。

Question：当 DAG 更新/导入失败时，日志在哪里可以指示是否发生导入失败以及确切的错误消息是什么？

目前，DAG解析日志将在$AIRFLOW_HOME/logs/EXECUTION_DATE/scheduler/DAG_FILE.py.log

Example:

假设我的 DAG 文件是example-dag.py其中包含以下内容，您可以注意到其中有一个拼写错误datetime import:

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import dattime   # <-- This Line has typo  


dag = DAG(
    dag_id='example_Dag',
    schedule_interval=None,
    start_date=datetime(2019, 2, 6),
)

t1 = BashOperator(
    task_id='print_date1',
    bash_command='sleep $[ ( $RANDOM % 30 )  + 1 ]s',
    dag=dag)

现在，如果您检查下面的日志$AIRFLOW_HOME/logs/scheduler/2021-04-07/example-dag.py.log where $AIRFLOW_HOME/logs是我设置的$AIRFLOW__LOGGING__BASE_LOG_FOLDER or [logging] base_log_folder in airflow.cfg (https://airflow.apache.org/docs/apache-airflow/2.0.1/configurations-ref.html#base-log-folder https://airflow.apache.org/docs/apache-airflow/2.0.1/configurations-ref.html#base-log-folder)

该文件应包含如下日志：

[2021-04-07 21:39:02,222] {scheduler_job.py:182} INFO - Started process (PID=686) to work on /files/dags/example-dag.py
[2021-04-07 21:39:02,230] {scheduler_job.py:633} INFO - Processing file /files/dags/example-dag.py for tasks to queue
[2021-04-07 21:39:02,233] {logging_mixin.py:104} INFO - [2021-04-07 21:39:02,233] {dagbag.py:451} INFO - Filling up the DagBag from /files/dags/example-dag.py
[2021-04-07 21:39:02,368] {logging_mixin.py:104} INFO - [2021-04-07 21:39:02,357] {dagbag.py:308} ERROR - Failed to import: /files/dags/example-dag.py
Traceback (most recent call last):
  File "/opt/airflow/airflow/models/dagbag.py", line 305, in _load_modules_from_file
    loader.exec_module(new_module)
  File "<frozen importlib._bootstrap_external>", line 678, in exec_module
  File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
  File "/files/dags/example-dag.py", line 3, in <module>
    from datetime import dattime
ImportError: cannot import name 'dattime'
[2021-04-07 21:39:02,380] {scheduler_job.py:645} WARNING - No viable dags retrieved from /files/dags/example-dag.py
[2021-04-07 21:39:02,407] {scheduler_job.py:190} INFO - Processing /files/dags/example-dag.py took 0.189 seconds

您将在 Web 服务器中看到如下错误：