使用IDE工具调试Airflow任务？

Question

我的 Airflow DAG 主要由 PythonOperators 组成，我想使用我的 Python IDE 调试工具来开发 python“内部”airflow。 - 我依赖 Airflow 的数据库连接器，我认为将其“移出”airflow 进行开发会很丑陋。

我已经使用Airflow有一段时间了，到目前为止仅通过CLI实现了开发和调试。这开始变得令人厌烦。

有谁知道设置 PyCharm 或其他 IDE 的好方法，使我能够在运行时使用 IDE 的调试工具集

airflow test ..

？

Answer 1

聚会可能有点晚了，但也一直在寻找解决方案。希望能够尽可能接近“生产模式”调试代码（因此无需进行测试等）。

找到了“Python 调试服务器”形式的解决方案。 它的工作方式相反：您的 IDE 进行监听，并从远程脚本到您的编辑器建立连接。

只需添加“Python 调试服务器”类型的新运行配置。 您将看到一个屏幕，告诉您远程

pip install pydevd-pycharm

。在同一页面上，您可以填写本地 IP 和调试器可用的端口以及可选的路径映射。

之后，只需将建议的 2 行代码添加到您希望调试会话开始的位置即可。

运行配置以激活侦听器，如果一切顺利，您的编辑器应该在到达

settrace

调用的位置后立即中断。

编辑/注意：如果您在编辑器中停止配置，airflow 将继续执行任务，请务必意识到这一点。

Answer 2

这可能有点像黑客，但我找到了一种设置 PyCharm 的方法：

• 使用

  which airflow

  到本地气流环境 - 在我的例子中只是一个 pipelinev
• 在 PyCharm 中添加新的运行配置
• 将 python“脚本路径”设置为所述气流脚本
• 设置参数来测试任务：

  test dag_x task_y 2019-11-19

这仅通过 SequentialExecutor 进行了验证，这可能很重要。

我必须为每个新的调试/开发任务更改运行配置中的测试参数，这很糟糕，但到目前为止，这对于在本地气流环境“内部”时设置断点和单步执行代码非常有用。

Answer 3

对于 VSCode，以下调试配置附加内置调试器

    {
        "name": "Airflow Test - Example",
        "type": "python",
        "request": "launch",
        "program": "`pyenv which airflow`",  // or path to airflow 
        "console": "integratedTerminal",
        "args": [ // exact formulation may depend on airflow 1.0 vs 2.0
            "test",
            "mydag",
            "mytask",
            "`date +%Y-%m-%dT00:00:00`", // current date 
            "-sd",
            "path/to/mydag" // providing the subdirectory makes this faster
        ]
    }

我假设其他 IDE 也有类似的配置

Answer 4

我调试

airflow test dag_id task_id

，使用 PyCharm 在 vagrant 机器上运行。即使您直接在本地主机上运行气流，您也应该能够使用相同的方法。

Pycharm 关于此主题的文档 应该向您展示如何创建适当的“Python 远程调试”配置。当您运行此配置时，它会等待您在某处添加的代码位（例如在您的操作员之一中）联系。然后就可以正常调试了，在 Pycharm 中设置断点。

Answer 5

如果您使用 docker compose 和 Airflow，Python 调试服务器的工作方式相同。定期启动容器，创建运行配置并在 docker 容器中安装所需的包（例如 webserver/scheduler）。
对我有用的 IDE 主机名是

host.docker.internal

以及任何未使用的端口。我还连接到容器并运行 DAG，如下所示：

python dags/your_dag.py

DAG 文件：

dag = ... # generate or create your dag
if __name__ == "__main__":
  import pydevd_pycharm
  pydevd_pycharm.settrace('host.docker.internal', port=9673, stdoutToServer=True, stderrToServer=True)
  dag.test()

不要忘记在运行配置中创建路径映射。