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)

标签： pythonairflow

Answer 1

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

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': '[e-mail]',
    '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)

结果是这样；