如何让Airflow的OracleStoredProcedureOperator在存储过程失败时触发任务失败？

核心原因 #

OracleStoredProcedureOperator依赖cx_Oracle调用存储过程，如果存储过程内部仅出现错误但未主动抛出异常（比如ORA-01403这类错误被静默处理），cx_Oracle不会向上传递错误，导致Airflow无法感知失败，进而标记任务为成功。

解决方案 #

1. 修改存储过程，主动抛出异常 #

在存储过程中添加异常处理块，将错误转为明确的应用级异常，确保cx_Oracle能捕获到：

CREATE OR REPLACE PROCEDURE schema.load_table.load_end(p_arg IN VARCHAR2)
IS
BEGIN
    -- 原存储过程逻辑
    ...
EXCEPTION
    WHEN NO_DATA_FOUND THEN
        -- 捕获ORA-01403并抛出自定义异常
        RAISE_APPLICATION_ERROR(-20001, '执行失败：未找到数据（ORA-01403）', TRUE);
    WHEN OTHERS THEN
        -- 捕获其他所有错误并抛出
        RAISE_APPLICATION_ERROR(-20002, '存储过程执行失败：' || SQLERRM, TRUE);
END;
/

当存储过程执行出错时，会主动抛出异常，cx_Oracle调用时会触发DatabaseError，Airflow会自动将任务标记为失败。

2. 自定义Operator增强异常捕获（可选） #

如果无法修改存储过程，或者需要更灵活的错误处理，可以扩展OracleStoredProcedureOperator，主动捕获数据库异常并转为Airflow异常：

from airflow.providers.oracle.operators.oracle import OracleStoredProcedureOperator
from airflow.exceptions import AirflowException
import cx_Oracle

class CustomOracleSPOperator(OracleStoredProcedureOperator):
    def execute(self, context):
        try:
            super().execute(context)
            # 若存储过程通过输出参数返回错误码，可在此处检查
            # 示例：假设最后一个参数是输出错误码
            # error_code = self.parameters[-1].getvalue()
            # if error_code != 0:
            #     raise AirflowException(f"存储过程执行失败，错误码：{error_code}")
        except cx_Oracle.DatabaseError as e:
            error_detail = e.args[0]
            raise AirflowException(
                f"存储过程执行失败：ORA-{error_detail.code} - {error_detail.message}"
            ) from e

使用自定义Operator替换原算子：

CustomOracleSPOperator(
    task_id=f"{table_id}_oracle_load_end_procedure",
    oracle_conn_id=oracle_conn_id,
    procedure="schema.load_table.load_end",
    parameters=[oracle_procedure_argument],
    do_xcom_push=True
)

3. 验证配置 #

确保Airflow的Oracle连接配置正确，且连接用户拥有足够权限执行存储过程并查看错误信息。

内容的提问来源于stack exchange，提问作者PipelineSurfer