使用 CLI 运行命令在 Airflow 中触发任务 [英] Triggering a task in Airflow with the CLI run command

问题描述

如何使用 CLI 中的 run 命令在 Airflow (v1.8.1) DAG 中触发特定任务?

How do I trigger a specific task in an Airflow (v1.8.1) DAG using the run command from the CLI?

DAG

dag = DAG(
    'test_run',
    default_args=default_args,
    schedule_interval=SCHEDULE_INTERVAL,
    catchup=False
)

one = BashOperator(
    task_id='one',
    bash_command='echo 1',
    dag=dag,
)

two = BashOperator(
    task_id='two',
    bash_command='echo 2',
    dag=dag,
)

one.set_downstream(two)

命令

airflow run -i -I -f test_run two 2018-08-21T18:59:51

我已经尝试了 -i、-I 和 -f 命令，但没有任何运气和错误.我也试过 -A 忽略所有依赖项，但失败告诉我命令中没有执行日期.

I've tried the -i, -I and -f commands without any luck and without error. I've also tried -A for ignoring all dependencies, but that fails telling me there is no execution date in the command.

如果我运行相同的命令但指定任务one，我可以看到在任务实例列表中成功显示，但我似乎无法完全运行任务two忽略任务one.

If I run the same command but specify task one, I can see that successfully show in the task instance list, but I cannot seem to run task two while completely ignoring task one.

尝试运行 two 时的实际输出.

Actual output when attempting to run two.

datauser@ubuntu-xenial:/home/vagrant$ AIRFLOW_HOME=~/code/dataprocessing/ /virtualenvs/dataprocessing/bin/airflow run -i -I -f test_run two 2018-08-21T18:59:51
/virtualenvs/dataprocessing/lib/python3.5/site-packages/psycopg2/__init__.py:144: UserWarning: The psycopg2 wheel package will be renamed from release 2.8; in order to keep installing from binary please use "pip install psycopg2-binary" instead. For details see: <http://initd.org/psycopg/docs/install.html#binary-install-from-pypi>.
  """)
[2018-08-21 19:11:51,538] {__init__.py:57} INFO - Using executor CeleryExecutor
[2018-08-21 19:11:51,624] {driver.py:120} INFO - Generating grammar tables from /usr/lib/python3.5/lib2to3/Grammar.txt
[2018-08-21 19:11:51,650] {driver.py:120} INFO - Generating grammar tables from /usr/lib/python3.5/lib2to3/PatternGrammar.txt
Sending to executor.
INFO:root:The key logs/test_run/two/2018-08-21T18:59:51 now contains 101162 bytes

深入研究日志，我可以看到:

And digging into the logs, I can see:

[2018-08-21 19:13:49,796] {base_task_runner.py:95} INFO - Subtask: [2018-08-21 19:13:49,795] {models.py:1122} INFO - Dependencies not met for <TaskInstance: test_run.two 2018-08-21 18:59:51 [None]>, dependency 'Trigger Rule' FAILED: Task's trigger rule 'all_success' requires all upstream tasks to have succeeded, but found 1 non-success(es). upstream_tasks_state={'successes': 0, 'upstream_failed': 0, 'done': 0, 'skipped': 0, 'failed': 0}, upstream_task_ids=['one']

但是 -I 标志不应该处理这个吗?

But should the -I flag not take care of this?

如果在没有 -I 标志的情况下运行，但使用 -i -f 标志，我会收到同样的错误.

If run without the -I flag, but with the -i -f flags, I receive the same error.

[2018-08-22 16:47:19,582] {base_task_runner.py:95} INFO - Subtask: [2018-08-22 16:47:19,581] {models.py:1122} INFO - Dependencies not met for <TaskInstance: test_run.two 2018-08-22 00:59:51 [None]>, dependency 'Trigger Rule' FAILED: Task's trigger rule 'all_success' requires all upstream tasks to have succeeded, but found 1 non-success(es). upstream_tasks_state={'upstream_failed': 0, 'successes': 0, 'done': 0, 'skipped': 0, 'failed': 0}, upstream_task_ids=['one']

推荐答案

这是 Airflow 版本 <= 1.8 中的一个错误.这在 Airflow 版本 >= 1.9 中确实有效.

This is a bug in Airflow version <= 1.8. This does work in Airflow version >= 1.9.

这篇关于使用 CLI 运行命令在 Airflow 中触发任务的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！