python - 在 Apache Airflow 中，有没有办法捕获 bash 命令的错误？

Question

在 Apache Airflow 中，是否可以捕获失败的 bash 命令生成的原始错误消息，而不是 Apache Airflow 生成的回溯错误，该错误告诉您该行失败，但不确切说明失败的原因？

Dag 中的示例行:

gsutil_rsync = BashOperator(
        task_id="task1",
        bash_command='gsutil rsync -r s3://bucket/ gs://bucket',
        dag=dag)

Answer 1

我使用 python 函数和 PythonOperator 编写了这个解决方案，并在 PythonOperator 中设置了 xcom_push=True。

import subprocess
from datetime import datetime

from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2018, 10, 15),
    'email': 'me@airflow.com',
    'email_on_failure': False,
    'retries': 1,
    'retry_delay': 1,
}


def run_bash():
    result = subprocess.run(['ls', '-l'], stdout=subprocess.PIPE)
    return result.stdout

run_bash()

with DAG('bash_dag', schedule_interval="@daily", default_args=default_args) as dag:
    start_brach = DummyOperator(task_id='start')

    gsutil_rsync_py = PythonOperator(
        task_id="task1",
        python_callable=run_bash,
        xcom_push=True,
        dag=dag)


    start_brach.set_downstream(gsutil_rsync_py)

结果是这样的；