exception - 在 Airflow 中传递异常

Question

我是 Apache Airflow 的新手，我计划使用 BashOperator 类运行 Python 和 R 脚本文件。我想了解 Exceptions 在两种情况下应该如何工作：
1. R 或 Python 脚本由于某种原因失败；或
2. R 或 Python 脚本已完成，但我希望在 DAG 继续执行下一个任务之前需要人工输入。

我有两个非常基本的问题：
1. 异常如何从 R 或 Python 脚本文件传递到 BashOperator 再到 DAG？例如，对 R 脚本文件的调用是否应该在 BashOperator 的 try 块中？
2. 如何传递自定义异常（警告？错误？），以便即使 R 或 Python 脚本成功完成，我也可以暂停 DAG 的执行？

我很感激您可以指出我的任何 Airflow 异常处理示例。

标签： exceptionairflow

Answer 1

忘记您正在使用任何 Airflow 运算符。

只需说您正在编写一个运行 R 或 Python 脚本的 Bash 脚本。

案例 1：如果 Python 脚本失败，Bash 脚本会失败：

set -e
python test_file.py

案例 2：即使 Python 脚本失败也能通过的 Bash 脚本：

python test_file.py || exit 0

现在只需在 BashOperator 中传递上述 bash 脚本之一。

基本上，BashOperator只需运行普通的 Bash 命令/脚本并传递脚本的日志和退出状态。在案例 1 中，您的 Airflow 任务也会失败，在案例 2 中，Airflow 任务也会失败。