pycharm 使用IDE工具执行气流任务?

deikduxw  于 12个月前  发布在  PyCharm
关注(0)|答案(5)|浏览(196)

我的Airflow DAG主要由PythonOperators组成,我想使用我的Python IDE调试工具来开发python“inside”airflow。- 我依赖于Airflow的数据库连接器,我认为将其“移出”Airflow进行开发是丑陋的。
我已经使用Airflow一段时间了,到目前为止只通过CLI实现了开发和调试。开始让人厌烦了。
有没有人知道一种很好的方法来设置PyCharm或其他IDE,使我能够在运行airflow test ..时使用IDE的调试工具集?

wqnecbli

wqnecbli1#

可能有点晚了,但也在寻找解决办法。希望能够调试代码尽可能接近“生产模式”(所以没有测试等)。
找到了一个解决方案,形式为“Python服务器”。它的工作原理是相反的:您的IDE会进行侦听,并建立从远程脚本到编辑器的连接。
只需添加一个新的运行配置类型“Python Server”。你会得到一个屏幕告诉你远程pip install pydevd-pycharm。在同一页中,您可以填写本地IP和调试器应该可用的端口以及可选的路径Map。
之后,只需将建议的2行代码添加到您希望开始调试会话的位置。
运行配置来激活监听器,如果一切正常,那么一旦到达settrace-调用的位置,编辑器就应该中断。

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

vmdwslir

vmdwslir2#

这可能有点像黑客,但我发现了一种设置PyCharm的方法:

  • 在本地气流环境中使用which airflow-在我的情况下,它只是一个pipenv
  • 在PyCharm中添加新的运行配置
  • 将python“脚本路径”设置为上述气流脚本
  • 设置测试任务的参数:test dag_x task_y 2019-11-19

这只使用SequentialExecutor进行了验证,这可能很重要。
我不得不为每个新的调试/开发任务更改运行配置中的测试参数,这很糟糕,但到目前为止,这对于设置断点和在本地气流环境中单步执行代码非常有用。

rfbsl7qr

rfbsl7qr3#

对于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也有类似的工作环境

carvr3hs

carvr3hs4#

我调试airflow test dag_id task_id,运行在一个流浪机器上,使用PyCharm。您应该能够使用相同的方法,即使您直接在localhost上运行airflow。
Pycharm关于这个主题的文档应该向你展示如何创建一个合适的“Python远程脚本”配置。当你运行这个配置时,它会等待你在某个地方添加的代码(例如在你的一个操作符中)。然后你就可以正常调试了,在Pycharm中设置断点。

o2rvlv0m

o2rvlv0m5#

如果你使用docker compose和Airflow,Python 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()

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

相关问题