You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@airflow.apache.org by Hongyi Wang <wh...@gmail.com> on 2022/01/28 19:45:57 UTC
[DISCUSSION] Specify tasks to skip when triggering DAG
Hello everyone,
I'd like to propose a new feature in Airflow -- allow users to specify
tasks to skip when trigger DAG run.
From our own experience, this feature can be very useful when doing
experiments, troubleshooting or re-running existing DAGs. And I
believe it can benefit many Airflow users.
To illustrate the use case, I am going to use this example below.
task-a ☐ -> task-b ☑ -> task-c ☐
Suppose we have a DAG containing 3 tasks. To troubleshoot "task-a" and
"task-c", I want to trigger a manual DAG run and skip "task-b" (so I
can save time & resource & focus on other two tasks). To do so, today
I have two options:
Option 1: Trigger DAG, then manually mark "task-b" as `SUCCESS`
Option 2: Remove "task-b" from my DAG, then trigger DAG
Neither of the options are great. Option 1 can be troublesome when DAG
is large, and there are multiple tasks I want to skip. Option 2
requires change in the DAG file, which is not convenient for just
troubleshooting.
Therefore, I would love to discuss how we can provide an easy way for
users to skip tasks when triggering DAG.
Things to consider are:
1) We should allow user to specify all tasks to skip at once when trigger DAG
2) We should retain the dependencies between non-skip tasks (in above
example, "task-c" won't start until "task-a" completes even if we
skipped "task-b")
3) We should mark skipped task as `SKIPPED` instead of `SUCCESS` to
make it more intuitive
4) The implementation should be easy, clean and low risk
Here is my proposed solution (tested locally):
Today, Airflow allow user to pass a JSON to the Dagrun as
{{dag_run.conf}} when triggering DAG. The idea is, before queuing task
instances that satisfies dependences, `scheduler_job.py` (after we
make some change) will filter task instances to skip based on
`dag_run.conf` user passes in (e.g. {"skip_tasks": ["task-b"]}), then
mark them as SKIPPED.
Things I would love to discuss:
- What do you think about this feature?
- What do you think about the proposed solution?
- Did I miss anything that you want to discuss?
- Is it necessary to introduce a new state (e.g. MANUAL_SKIPPED) to
differentiate SKIPPED?
Howie
Re: [DISCUSSION] Specify tasks to skip when triggering DAG
Posted by Jarek Potiuk <ja...@potiuk.com>.
Not that I am unhappy :). I just think it's quite a bit ambiguous to
treat "success" and "pretending success" without some indication.
But I won't oppose it if others also want this.
This is not a strong "no". Eventually it is more the question to the
users that might or might not get confused.
It's more "why skipped is not enough". Is this possible to see some
real cases that cannot be done with "skip" and existing rules Hongyi?
J.
On Wed, Feb 16, 2022 at 12:04 AM Daniel Standish
<da...@astronomer.io.invalid> wrote:
>
> Thinking more, equivalently, when raising AirflowSkipException, we could optionally include a payload, which would determine the "final" task state.
>
> I suppose neither of those would really make Jarek happier because for both of these the final state doesn't convey whether it did work or not.
>
> What you are suggesting -- a new state -- actually does seem to address that concern. You are suggesting that we add an entirely new state, which would be rendered in a different way and therefore would be distinguishable from both success and skipped.
>
> But personally I think I'd probably rather see us add an AirflowSucessException and if you want to do that (with the resulting ambiguity pointed out by Jarek) then that's up to you. Because it's very simple and it achieves your main goal.
>
> My concern with more states is that we already have too many states (note how much horizontal space they take up on the UI) and they are muddled and overlapping (IMO they want to be split into different categories) and the goal can be achieved without a new one.
Re: [DISCUSSION] Specify tasks to skip when triggering DAG
Posted by Daniel Standish <da...@astronomer.io.INVALID>.
Thinking more, equivalently, when raising AirflowSkipException, we could
optionally include a payload, which would determine the "final" task
state.
I suppose neither of those would really make Jarek happier because for both
of these the final state doesn't convey whether it did work or not.
What you are suggesting -- a new state -- actually does seem to address
that concern. You are suggesting that we add an entirely new state, which
would be rendered in a different way and therefore would be distinguishable
from both success and skipped.
But personally I think I'd probably rather see us add an
AirflowSucessException and if you want to do that (with the resulting
ambiguity pointed out by Jarek) then that's up to you. Because it's very
simple and it achieves your main goal.
My concern with more states is that we already have too many states (note
how much horizontal space they take up on the UI) and they are muddled and
overlapping (IMO they want to be split into different categories) and the
goal can be achieved without a new one.
Re: [DISCUSSION] Specify tasks to skip when triggering DAG
Posted by Daniel Standish <da...@astronomer.io.INVALID>.
One thing you could do is add an `AirflowSuccessException`. Similar to
AirflowSkipException, if raised task would immediately stop and be marked
as successful. Then in your implementation of the pre-execute callable,
you can choose whether you want it to be "skip" or "skip success".
Re: [DISCUSSION] Specify tasks to skip when triggering DAG
Posted by Jarek Potiuk <ja...@potiuk.com>.
I think marking a task as a success where it has not been executed is
really an anti-pattern after thinking a bit. \
I really think that if tasks should be successful in some conditions
where they should not actually "do stuff" is the logic of the task,
not airflow behaviour around it. This does not add value for the user
- mostly because you cannot distinguish the state in the UI. This will
be confusing like crazy when you have DAGs which will look the same if
the task was effectively "skipped" or "executed" - this is abusing
rather than using Airflow.
J.
On Mon, Feb 14, 2022 at 2:11 AM Hongyi Wang <wh...@gmail.com> wrote:
>
> Hi Daniel and Jarek,
>
> After discussing with XD and the rest of our team members, we decided to move forward with the `pre_execute` solution. Thank you @Deniel for raising the idea. Thank you @Jarek for explaining the tradeoffs in detail.
>
> Besides "how to skip tasks", another thing I'd love to discuss is the "skip state to set".
>
> | So the skipped task needs to have a State identical to "Success", other than the name. Hence a new state may be needed
>
> When skipping a task, (in `pre_execute`) we can either choose to mark it as "SUCCESS" or "SKIPPED". To differentiate skipped tasks from successful ones, "SKIPPED" is preferred, however, it may not work well with some trigger_rules like `all_success`. Therefore, I am thinking to introduce a new state e.g. "SKIPPED_AS_SUCCESS", which is different from "SUCCESS", but will be treated the same as "SUCCESS" when evaluating dependency rules.
>
> WYT?
>
> Howie
>
> On Sat, Feb 5, 2022 at 8:10 AM Jarek Potiuk <ja...@potiuk.com> wrote:
>>
>> I thought a bit about it and there is one more thing that I do not really like about the approach when you specify the task to skip in the dag run conf.
>>
>> Currently the "logic" of the dag run conf is exclusively "dag specific" and only used at "task execution" time (mind - it is not used in "dag scheduling time" at all. Currently "Airflow Core" does not have to know anything about it and can transparently pass it to the task without parsing it. And it's the logic inside the DAG that should "understand" it. There is not a single other use case where dag_run conf is used and parsed by scheduler.
>> Trying to make "dag_run.conf" to be used either in "Airflow Core" or in "DAG logic" depending on the parameter, is IMHO very wrong because it is mixing independent concerns.
>>
>> Currently things are simple:
>>
>> * dag structure is defined by DAG parsing (this is the only thing Scheduler cares)
>> * task execution uses the "static" structure defined by DAG parsing and uses dag_run conf to implement logic to react to different parameters
>>
>> When you look at this from that perspective - the proposal is a weird mixture of both. And if we do it, in the future it **might** prevent us from doing a number of optimizations. For example dag_run_conf does not have to be parsed at all when the scheduler does its job. The only place dag_run conf is parsed is only at the moment the task is executed.
>>
>> For me this is a huge '-1' of the solution.
>>
>> Coming back XD to your example - I think what you are describing there is a particular DAG requirement, not something that Airflow core should take care about. Also - no matter how the task will be skipped (whether in pre-execute or by scheduler, if you want to "Skip" B2 but behave as if it "succeeded" you have exactly the same problems with triggering rules. It completely does not matter if you "skip" by throwing an exception or "Skip" by airflow. What you are really talking about it is not to "skip" certain tasks but rally about "pretending they succeeded". This is an entirely different thing than that we started the discussion with :). We were talking about "skipping" the task - which in Airflow is a different state than "Succeeded".
>>
>> And if you really want to "pretend success" you don't even have to (or even should) use 'pre_execute' for it. There are better ways already.
>>
>> If you are using the "modern" way of writing tasks (which I assume is the case for anyone who does custom work like that) I imagine your B2 task should be written this way (writing from memory so it might not compile):
>>
>> @task
>> def task_b2(context: 'Context'):
>> if context.dag_run.conf.get("should_task2_pretend_it_succeeded"):
>> return
>> # do stuff (remember to use hooks instead of the operators in the @task decorated tasks to use multiple Airflow providers)
>>
>> IMHO - this is sooooooo much more "Modern Airflow way" of approaching the problem:
>>
>> * it's very simple and follows the "modern" way of writing Airflow tasks (see also my talk about it from this week NYC Airflow Meetup https://photos.app.goo.gl/ycCLpCmQ9fiTDNNDA)
>> * the logic of "pretending" is exactly there next to the code that otherwise should be run and it's very obvious what happens (otherwise you have to get a mental jump from DAG parameters passed through scheduler and task execution)
>> * it's extremely explicit (https://www.python.org/dev/peps/pep-0020/) - "explicit is better than implicit"
>> * dag_run configuration does not have to be understood by scheduler
>> * your DAG structure does not have to be "complicated" as the task will end up in "success" state
>> * this is imperative, not declarative, and DAG writer decides how "limited" the conditions are - Airflow does not limit you in any way and the condition can be arbitrary complex or simple (and take into account other parameters, dag_run, logical dates, actual dates of execution and basically anything else.
>>
>> From all the comments I saw - I am pretty convinced this is a much better way to approach it.
>>
>> BTW. There is indeed one "potential" pro of the "scheduler-based" skip calculation. It can provide a little optimization - the task does not need to be run at all to make the decision in this case (similarly what we do with DummyTasks now). But IMHO, this is totally offset by the fact that the scheduler would have to parse and analyse the dag_run conf to make the decisions - overall, it could be even slower in a number of cases. The scheduler loop is pretty critical and any extra logic and parsing there might have huge, unforeseen initial impact and we should only add any logic there if we are absolutely sure all the potential performance impact is well analyzed and understood.
>>
>>
>> Giorgio Zoppi,
>>
>> XCom being "small" is also a thing of the past (if you use custom XCom backends) - see the same talk of mine https://photos.app.goo.gl/ycCLpCmQ9fiTDNNDA ) but also https://airflow.apache.org/docs/apache-airflow/stable/concepts/xcoms.html?highlight=custom%20xcom#custom-backends and this nice article from Astronomer https://www.astronomer.io/guides/custom-xcom-backends
>>
>> J.
>>
>>
>>
>> On Sat, Feb 5, 2022 at 4:22 PM Giorgio Zoppi <gi...@gmail.com> wrote:
>>>
>>> Hey Jurek,
>>> Just a question about the future development, is the XComm backend replaceable now? The real power of Airflow is the 'defacto' the glue between different ways of mangling data, such as Python is the glue when you need to implement things a lower level, ie. Altair Simulation software is written for critical parts in C++ and binded at upper level in Python. Same we can state for Tensorflow, numpy. About my question: it would be nice to have as XComm backend a queue like redis or a non structured database for allowing scaling but i am not sure
>>> it it makes sense since XComm is just for short messages.
>>> Best Regards,
>>> Giorgio
>>>
>>>
>>>
Re: [DISCUSSION] Specify tasks to skip when triggering DAG
Posted by Hongyi Wang <wh...@gmail.com>.
Hi Daniel and Jarek,
After discussing with XD and the rest of our team members, we decided to
move forward with the `pre_execute` solution. Thank you @Deniel for raising
the idea. Thank you @Jarek for explaining the tradeoffs in detail.
Besides "how to skip tasks", another thing I'd love to discuss is the
"skip state to set".
| So the skipped task needs to have a State identical to "Success", other
than the name. Hence a new state may be needed
When skipping a task, (in `pre_execute`) we can either choose to mark it as
"SUCCESS" or "SKIPPED". To differentiate skipped tasks from successful
ones, "SKIPPED" is preferred, however, it may not work well with some
trigger_rules like `all_success`. Therefore, I am thinking to introduce a
new state e.g. "SKIPPED_AS_SUCCESS", which is different from "SUCCESS", but
will be treated the same as "SUCCESS" when evaluating dependency rules.
WYT?
Howie
On Sat, Feb 5, 2022 at 8:10 AM Jarek Potiuk <ja...@potiuk.com> wrote:
> I thought a bit about it and there is one more thing that I do not really
> like about the approach when you specify the task to skip in the dag run
> conf.
>
> Currently the "logic" of the dag run conf is exclusively "dag specific"
> and only used at "task execution" time (mind - it is not used in "dag
> scheduling time" at all. Currently "Airflow Core" does not have to know
> anything about it and can transparently pass it to the task without parsing
> it. And it's the logic inside the DAG that should "understand" it. There is
> not a single other use case where dag_run conf is used and parsed by
> scheduler.
> Trying to make "dag_run.conf" to be used either in "Airflow Core" or in
> "DAG logic" depending on the parameter, is IMHO very wrong because it is
> mixing independent concerns.
>
> Currently things are simple:
>
> * dag structure is defined by DAG parsing (this is the only thing
> Scheduler cares)
> * task execution uses the "static" structure defined by DAG parsing and
> uses dag_run conf to implement logic to react to different parameters
>
> When you look at this from that perspective - the proposal is a weird
> mixture of both. And if we do it, in the future it **might** prevent us
> from doing a number of optimizations. For example dag_run_conf does not
> have to be parsed at all when the scheduler does its job. The only place
> dag_run conf is parsed is only at the moment the task is executed.
>
> For me this is a huge '-1' of the solution.
>
> Coming back XD to your example - I think what you are describing there is
> a particular DAG requirement, not something that Airflow core should take
> care about. Also - no matter how the task will be skipped (whether in
> pre-execute or by scheduler, if you want to "Skip" B2 but behave as if it
> "succeeded" you have exactly the same problems with triggering rules. It
> completely does not matter if you "skip" by throwing an exception or "Skip"
> by airflow. What you are really talking about it is not to "skip" certain
> tasks but rally about "pretending they succeeded". This is an entirely
> different thing than that we started the discussion with :). We were
> talking about "skipping" the task - which in Airflow is a different state
> than "Succeeded".
>
> And if you really want to "pretend success" you don't even have to (or
> even should) use 'pre_execute' for it. There are better ways already.
>
> If you are using the "modern" way of writing tasks (which I assume is the
> case for anyone who does custom work like that) I imagine your B2 task
> should be written this way (writing from memory so it might not compile):
>
> @task
> def task_b2(context: 'Context'):
> if context.dag_run.conf.get("should_task2_pretend_it_succeeded"):
> return
> # do stuff (remember to use hooks instead of the operators in the @task
> decorated tasks to use multiple Airflow providers)
>
> IMHO - this is sooooooo much more "Modern Airflow way" of approaching the
> problem:
>
> * it's very simple and follows the "modern" way of writing Airflow tasks
> (see also my talk about it from this week NYC Airflow Meetup
> https://photos.app.goo.gl/ycCLpCmQ9fiTDNNDA)
> * the logic of "pretending" is exactly there next to the code that
> otherwise should be run and it's very obvious what happens (otherwise you
> have to get a mental jump from DAG parameters passed through scheduler and
> task execution)
> * it's extremely explicit (https://www.python.org/dev/peps/pep-0020/) -
> "explicit is better than implicit"
> * dag_run configuration does not have to be understood by scheduler
> * your DAG structure does not have to be "complicated" as the task will
> end up in "success" state
> * this is imperative, not declarative, and DAG writer decides how
> "limited" the conditions are - Airflow does not limit you in any way and
> the condition can be arbitrary complex or simple (and take into account
> other parameters, dag_run, logical dates, actual dates of execution and
> basically anything else.
>
> From all the comments I saw - I am pretty convinced this is a much better
> way to approach it.
>
> BTW. There is indeed one "potential" pro of the "scheduler-based" skip
> calculation. It can provide a little optimization - the task does not need
> to be run at all to make the decision in this case (similarly what we do
> with DummyTasks now). But IMHO, this is totally offset by the fact that the
> scheduler would have to parse and analyse the dag_run conf to make the
> decisions - overall, it could be even slower in a number of cases. The
> scheduler loop is pretty critical and any extra logic and parsing there
> might have huge, unforeseen initial impact and we should only add any logic
> there if we are absolutely sure all the potential performance impact is
> well analyzed and understood.
>
>
> Giorgio Zoppi,
>
> XCom being "small" is also a thing of the past (if you use custom XCom
> backends) - see the same talk of mine
> https://photos.app.goo.gl/ycCLpCmQ9fiTDNNDA ) but also
> https://airflow.apache.org/docs/apache-airflow/stable/concepts/xcoms.html?highlight=custom%20xcom#custom-backends
> and this nice article from Astronomer
> https://www.astronomer.io/guides/custom-xcom-backends
>
> J.
>
>
>
> On Sat, Feb 5, 2022 at 4:22 PM Giorgio Zoppi <gi...@gmail.com>
> wrote:
>
>> Hey Jurek,
>> Just a question about the future development, is the XComm backend
>> replaceable now? The real power of Airflow is the 'defacto' the glue
>> between different ways of mangling data, such as Python is the glue when
>> you need to implement things a lower level, ie. Altair Simulation software
>> is written for critical parts in C++ and binded at upper level in Python.
>> Same we can state for Tensorflow, numpy. About my question: it would be
>> nice to have as XComm backend a queue like redis or a non structured
>> database for allowing scaling but i am not sure
>> it it makes sense since XComm is just for short messages.
>> Best Regards,
>> Giorgio
>>
>>
>>
>>
Re: [DISCUSSION] Specify tasks to skip when triggering DAG
Posted by Jarek Potiuk <ja...@potiuk.com>.
I thought a bit about it and there is one more thing that I do not really
like about the approach when you specify the task to skip in the dag run
conf.
Currently the "logic" of the dag run conf is exclusively "dag specific" and
only used at "task execution" time (mind - it is not used in "dag
scheduling time" at all. Currently "Airflow Core" does not have to know
anything about it and can transparently pass it to the task without parsing
it. And it's the logic inside the DAG that should "understand" it. There is
not a single other use case where dag_run conf is used and parsed by
scheduler.
Trying to make "dag_run.conf" to be used either in "Airflow Core" or in
"DAG logic" depending on the parameter, is IMHO very wrong because it is
mixing independent concerns.
Currently things are simple:
* dag structure is defined by DAG parsing (this is the only thing Scheduler
cares)
* task execution uses the "static" structure defined by DAG parsing and
uses dag_run conf to implement logic to react to different parameters
When you look at this from that perspective - the proposal is a weird
mixture of both. And if we do it, in the future it **might** prevent us
from doing a number of optimizations. For example dag_run_conf does not
have to be parsed at all when the scheduler does its job. The only place
dag_run conf is parsed is only at the moment the task is executed.
For me this is a huge '-1' of the solution.
Coming back XD to your example - I think what you are describing there is a
particular DAG requirement, not something that Airflow core should take
care about. Also - no matter how the task will be skipped (whether in
pre-execute or by scheduler, if you want to "Skip" B2 but behave as if it
"succeeded" you have exactly the same problems with triggering rules. It
completely does not matter if you "skip" by throwing an exception or "Skip"
by airflow. What you are really talking about it is not to "skip" certain
tasks but rally about "pretending they succeeded". This is an entirely
different thing than that we started the discussion with :). We were
talking about "skipping" the task - which in Airflow is a different state
than "Succeeded".
And if you really want to "pretend success" you don't even have to (or even
should) use 'pre_execute' for it. There are better ways already.
If you are using the "modern" way of writing tasks (which I assume is the
case for anyone who does custom work like that) I imagine your B2 task
should be written this way (writing from memory so it might not compile):
@task
def task_b2(context: 'Context'):
if context.dag_run.conf.get("should_task2_pretend_it_succeeded"):
return
# do stuff (remember to use hooks instead of the operators in the @task
decorated tasks to use multiple Airflow providers)
IMHO - this is sooooooo much more "Modern Airflow way" of approaching the
problem:
* it's very simple and follows the "modern" way of writing Airflow tasks
(see also my talk about it from this week NYC Airflow Meetup
https://photos.app.goo.gl/ycCLpCmQ9fiTDNNDA)
* the logic of "pretending" is exactly there next to the code that
otherwise should be run and it's very obvious what happens (otherwise you
have to get a mental jump from DAG parameters passed through scheduler and
task execution)
* it's extremely explicit (https://www.python.org/dev/peps/pep-0020/) -
"explicit is better than implicit"
* dag_run configuration does not have to be understood by scheduler
* your DAG structure does not have to be "complicated" as the task will
end up in "success" state
* this is imperative, not declarative, and DAG writer decides how
"limited" the conditions are - Airflow does not limit you in any way and
the condition can be arbitrary complex or simple (and take into account
other parameters, dag_run, logical dates, actual dates of execution and
basically anything else.
From all the comments I saw - I am pretty convinced this is a much better
way to approach it.
BTW. There is indeed one "potential" pro of the "scheduler-based" skip
calculation. It can provide a little optimization - the task does not need
to be run at all to make the decision in this case (similarly what we do
with DummyTasks now). But IMHO, this is totally offset by the fact that the
scheduler would have to parse and analyse the dag_run conf to make the
decisions - overall, it could be even slower in a number of cases. The
scheduler loop is pretty critical and any extra logic and parsing there
might have huge, unforeseen initial impact and we should only add any logic
there if we are absolutely sure all the potential performance impact is
well analyzed and understood.
Giorgio Zoppi,
XCom being "small" is also a thing of the past (if you use custom XCom
backends) - see the same talk of mine
https://photos.app.goo.gl/ycCLpCmQ9fiTDNNDA ) but also
https://airflow.apache.org/docs/apache-airflow/stable/concepts/xcoms.html?highlight=custom%20xcom#custom-backends
and this nice article from Astronomer
https://www.astronomer.io/guides/custom-xcom-backends
J.
On Sat, Feb 5, 2022 at 4:22 PM Giorgio Zoppi <gi...@gmail.com>
wrote:
> Hey Jurek,
> Just a question about the future development, is the XComm backend
> replaceable now? The real power of Airflow is the 'defacto' the glue
> between different ways of mangling data, such as Python is the glue when
> you need to implement things a lower level, ie. Altair Simulation software
> is written for critical parts in C++ and binded at upper level in Python.
> Same we can state for Tensorflow, numpy. About my question: it would be
> nice to have as XComm backend a queue like redis or a non structured
> database for allowing scaling but i am not sure
> it it makes sense since XComm is just for short messages.
> Best Regards,
> Giorgio
>
>
>
>
Re: [DISCUSSION] Specify tasks to skip when triggering DAG
Posted by Giorgio Zoppi <gi...@gmail.com>.
Hey Jurek,
Just a question about the future development, is the XComm backend
replaceable now? The real power of Airflow is the 'defacto' the glue
between different ways of mangling data, such as Python is the glue when
you need to implement things a lower level, ie. Altair Simulation software
is written for critical parts in C++ and binded at upper level in Python.
Same we can state for Tensorflow, numpy. About my question: it would be
nice to have as XComm backend a queue like redis or a non structured
database for allowing scaling but i am not sure
it it makes sense since XComm is just for short messages.
Best Regards,
Giorgio
Re: [DISCUSSION] Specify tasks to skip when triggering DAG
Posted by Daniel Standish <da...@astronomer.io.INVALID>.
>
> [Cons]
> 1) Not scalable / Inconvenient
> To make a task skippable, one needs to modify existing DAG (to set
> `pre_execute`). It seems not difficult, but when your Airflow host a
> thousand DAGs own by different teams/users, it can be challenging.
You can use task_policy to apply (or chain, i would think) pre_execute
callables globally on your cluster.
https://airflow.apache.org/docs/apache-airflow/stable/concepts/cluster-policies.html
So the skipped task needs to have a State identical to "Success", other
> than the name. Hence a new state may be needed for this feature Howie is
> proposing.
> This is also one of the reasons why the *pre_execute* method may not
> fully resolve the question, because marking the skipped task as "*Skipped*"
> may lead to something we don't expect (marking it as "*Success*" is also
> a bad practice, because it will be a confusion when user checks the history
> later).
>
Yes, our states and trigger rules might need to become more expressive.
But it could still be done from pre_execute because we could have an
`AirflowSkipSuccess` exception or something like that. (not that this is
the only reasonable solution)
*A note on states*
This touches directly on a related thread that I have been thinking about a
lot of late.
We might want to separate "states" from "resolutions". I.e. running /
queued / scheduled / done could be thought of as "execution states" whereas
skipped / failed / success / upstream failed might be better represented as
"resolutions". Similarly when a task is paused (e.g. deferred, up for
reschedule) we might track those reasons separately. This came up recently
with a bug where start_date kept being updated with a reschedule-mode
sensor because, at the time when start_date is updated, the task no
longer knows it was "up for reschedule" . It's possible that refactoring
"states" could give us better knowledge about what's going on with the task
when we have to make decisions like "should we update the start date".
Re: [DISCUSSION] Specify tasks to skip when triggering DAG
Posted by Xiaodong Deng <xd...@apache.org>.
Hi everyone,
[DISCLAIMER first: I'm working in the same team as Howie]
Many thanks for the discussion. Very insightful!
I have two points to share and would like to hear how you think of them:
*# 1. Potential cons of `pre_execute` method*
I do agree it's a nice idea and is almost perfect for most Airflow users
(who author the workflows by writing DAGs, via writing Python, very
flexible). But I see increasing use cases where people create DAGs through
templating (including our teams), which makes it a little bit less flexible.
I get the point shared by Jarek, "*You can write common function in the
shared code that will be used across your teams*". Say we create a common
function for this "skip" purpose, what if the user already has another
*pre_execute* callable created for other purposes, then she/he will need to
modify the callable further. I cannot judge if this is a big or small
hassle for other Airflow users, but I think it will be a cool built-in
feature if users can skip certain tasks for a DagRun without having to make
any change to their DAGs.
*# 2. State and Trigger Rule (more important point to consider)*
I'm really happy that folks are thinking from the *State* and *Trigger Rule*
aspect. That's actually what came to my mind first when we discussed this
internally.
As James asked, "Skipped" state may cause unexpected side-effects. IMHO The
solution Jarek shared (using different trigger_rule, like "all_done") may
not thoroughly resolve the concern, especially for a more complicated DAG,
for example the one in the image below (if the image doesn't display,
please check it at https://postimg.cc/cK837ydy)
[image: Untitled.drawio (2).png]
We only want to skip task B2 (in ad-hoc manually-triggered runs). Our
target includes to retain the dependencies and the whole thing should work
like B2 succeeded.
No matter we apply "*all_done*" or "*non_failed*" for Task C, it will
always work differently from what we may really expect (we may expect Task
C to still behave like with default trigger rule "*all_success*")
So the skipped task needs to have a State identical to "Success", other
than the name. Hence a new state may be needed for this feature Howie is
proposing.
This is also one of the reasons why the *pre_execute* method may not fully
resolve the question, because marking the skipped task as "*Skipped*" may
lead to something we don't expect (marking it as "*Success*" is also a bad
practice, because it will be a confusion when user checks the history
later).
Of course I may have missed something or I may be wrong in some of the
analysis. Please help point out, and more discussions around this would be
greatly appreciated ;-) Many thanks.
Regards,
XD
On Sat, Feb 5, 2022 at 11:03 AM Jarek Potiuk <ja...@potiuk.com> wrote:
> Airflow is really a "pythonic" framework. The basic premise of Airflow is
> that everything "DAG" is not declarative but Imperative. DAG definition,
> Task code, decorated python callables as a "new" way of writing tasks, heck
> even triggers and more recently timetables are "python imperative code".
> Similarly - Custom Secret, XCom Backends are very little "configurable" -
> if you need to change their behaviour, you .... write code rather than (for
> example) provide mapping of variables. And that's a deliberate choice -
> which is different from many other frameworks. If anything - we take away
> "declarative" ways rather than add them. This is the power of Airflow,
> rather than adding more "limited" options. we add ways to extend
> Airflow capabilities by writing small snippets of code rather than
> "declaring" how existing code should behave.
>
> In this context I thin neither of your cons hold:
>
> 1) it is fully scalable. You can write common function in the shared code
> that will be used across your teams
> 2) this is how you usually extend airflow, your admins are already
> familiar with extending Airflow with writing small pieces of code, so it
> does not cost too much extra
> 3) No abuse. This is one of the million ways of using pre-execute and a
> very cool one. We've added experimentally "pre_execute" as callable not
> that long time ago and the more I see cases like it the more I think it was
> a good idea
>
> So I do not see it as a workaround at all. I see it as a regular "feature"
> of Airflow.
>
> Just my thoughts.
>
> J.
>
>
>
> On Sat, Feb 5, 2022 at 8:35 AM Hongyi Wang <wh...@gmail.com> wrote:
>
>> Hi Daniel,
>>
>> Thank you for sharing the `pre_execute` idea. It is a smart idea, and I
>> do believe it would work. But to me, it is more as a walkaround instead of
>> an universal solution. To explain, I would love to discuss the tradeoffs.
>>
>> [Pros]
>> The solution I proposed requires change in either Airflow Scheduler or
>> Executor. In comparison, the `pre_execute` solution doesn't require any
>> change, which avoids introducing additional risk/complexity.
>>
>> [Cons]
>> 1) Not scalable / Inconvenient
>> To make a task skippable, one needs to modify existing DAG (to set
>> `pre_execute`). It seems not difficult, but when your Airflow host a
>> thousand DAGs own by different teams/users, it can be challenging.
>>
>> 2) May cost additional time and resources
>> A task won’t be skipped until it reaches `pre_execute()`. It is okay when
>> using LocalExecutor/SequentialExecutor; however, when using
>> KubernetesExecutor, a k8s worker pod will always be created for “task to
>> skip”. Ideally, tasks are skipped during "scheduling".
>>
>> 3) May abuse `pre_execute`
>> Instead of asking developers to set `pre_execute=skip_if_specified` so
>> the task becomes skippable. Wouldn’t it be better if we make it (task is
>> skippable) an out-of-box feature, meanwhile `pre_execute` can serve its
>> original purpose?
>>
>> Howie
>>
>> On Fri, Feb 4, 2022 at 7:41 AM James Coder <jc...@gmail.com> wrote:
>>
>>> Thanks Jarek, yeah, I wasn't thinking about the trigger rules, hazards
>>> of emailing before coffee.
>>>
>>> On Fri, Feb 4, 2022 at 10:32 AM Jarek Potiuk <ja...@potiuk.com> wrote:
>>>
>>>> Thanks Daniel, this something I have been thinking about for a while.
>>>>> One use case I have is for a dag only gets run on an ad hoc basis, for any
>>>>> combination of a subset of tasks (I recognize this is a rather non-standard
>>>>> use of airflow). One question though, if you specified skip_list=[“task b”]
>>>>> wouldn’t the scheduler skip task c as well?
>>>>>
>>>>
>>>> Depending on the triggering rule you choose.
>>>>
>>>> One of the effects of the default "all_success" rule is that the "skip"
>>>> state propagates when the rule is used.
>>>> If you use "all_done", it won't be skipped (but failure will also be
>>>> counted as success) and when you use "none_failed" it will propagate the
>>>> "failure" of B to C as well (but "skip" will not be propagated)..
>>>>
>>>> Generally speaking when you expect some task to be skippable you should
>>>> design your DAG to account for it.
>>>>
>>>> You can see more about the rules in this really informative post by
>>>> Marc Lamberti
>>>> https://marclamberti.com/blog/airflow-trigger-rules-all-you-need-to-know/
>>>>
>>>> Whenever I can't remember how those rules work, I keep on coming back
>>>> to the post.
>>>>
>>>>
>>>>
>>>>> James Coder
>>>>> ------------------------------
>>>>> *From:* Daniel Standish <da...@astronomer.io.INVALID>
>>>>> *Sent:* Friday, February 4, 2022 1:41:12 AM
>>>>> *To:* dev@airflow.apache.org <de...@airflow.apache.org>
>>>>> *Subject:* Re: [DISCUSSION] Specify tasks to skip when triggering DAG
>>>>>
>>>>> That skip func had a typo (conf where it should have been context)...
>>>>>
>>>>> this is more likely to work:
>>>>>
>>>>> def skip_if_specified(context):
>>>>> if not context:
>>>>> return
>>>>> dr = context.get('dag_run')
>>>>> ti = context.get('task_instance')
>>>>> if not (dr and ti):
>>>>> return
>>>>> conf = dr.conf
>>>>> if not conf:
>>>>> return
>>>>> skip_list = conf.get('skip_list', [])
>>>>> if ti.task_id in skip_list:
>>>>> raise AirflowSkipException()
>>>>>
>>>>> Apologies for the spam.
>>>>>
>>>>
Re: [DISCUSSION] Specify tasks to skip when triggering DAG
Posted by Jarek Potiuk <ja...@potiuk.com>.
Airflow is really a "pythonic" framework. The basic premise of Airflow is
that everything "DAG" is not declarative but Imperative. DAG definition,
Task code, decorated python callables as a "new" way of writing tasks, heck
even triggers and more recently timetables are "python imperative code".
Similarly - Custom Secret, XCom Backends are very little "configurable" -
if you need to change their behaviour, you .... write code rather than (for
example) provide mapping of variables. And that's a deliberate choice -
which is different from many other frameworks. If anything - we take away
"declarative" ways rather than add them. This is the power of Airflow,
rather than adding more "limited" options. we add ways to extend
Airflow capabilities by writing small snippets of code rather than
"declaring" how existing code should behave.
In this context I thin neither of your cons hold:
1) it is fully scalable. You can write common function in the shared code
that will be used across your teams
2) this is how you usually extend airflow, your admins are already familiar
with extending Airflow with writing small pieces of code, so it does not
cost too much extra
3) No abuse. This is one of the million ways of using pre-execute and a
very cool one. We've added experimentally "pre_execute" as callable not
that long time ago and the more I see cases like it the more I think it was
a good idea
So I do not see it as a workaround at all. I see it as a regular "feature"
of Airflow.
Just my thoughts.
J.
On Sat, Feb 5, 2022 at 8:35 AM Hongyi Wang <wh...@gmail.com> wrote:
> Hi Daniel,
>
> Thank you for sharing the `pre_execute` idea. It is a smart idea, and I do
> believe it would work. But to me, it is more as a walkaround instead of an
> universal solution. To explain, I would love to discuss the tradeoffs.
>
> [Pros]
> The solution I proposed requires change in either Airflow Scheduler or
> Executor. In comparison, the `pre_execute` solution doesn't require any
> change, which avoids introducing additional risk/complexity.
>
> [Cons]
> 1) Not scalable / Inconvenient
> To make a task skippable, one needs to modify existing DAG (to set
> `pre_execute`). It seems not difficult, but when your Airflow host a
> thousand DAGs own by different teams/users, it can be challenging.
>
> 2) May cost additional time and resources
> A task won’t be skipped until it reaches `pre_execute()`. It is okay when
> using LocalExecutor/SequentialExecutor; however, when using
> KubernetesExecutor, a k8s worker pod will always be created for “task to
> skip”. Ideally, tasks are skipped during "scheduling".
>
> 3) May abuse `pre_execute`
> Instead of asking developers to set `pre_execute=skip_if_specified` so the
> task becomes skippable. Wouldn’t it be better if we make it (task is
> skippable) an out-of-box feature, meanwhile `pre_execute` can serve its
> original purpose?
>
> Howie
>
> On Fri, Feb 4, 2022 at 7:41 AM James Coder <jc...@gmail.com> wrote:
>
>> Thanks Jarek, yeah, I wasn't thinking about the trigger rules, hazards of
>> emailing before coffee.
>>
>> On Fri, Feb 4, 2022 at 10:32 AM Jarek Potiuk <ja...@potiuk.com> wrote:
>>
>>> Thanks Daniel, this something I have been thinking about for a while.
>>>> One use case I have is for a dag only gets run on an ad hoc basis, for any
>>>> combination of a subset of tasks (I recognize this is a rather non-standard
>>>> use of airflow). One question though, if you specified skip_list=[“task b”]
>>>> wouldn’t the scheduler skip task c as well?
>>>>
>>>
>>> Depending on the triggering rule you choose.
>>>
>>> One of the effects of the default "all_success" rule is that the "skip"
>>> state propagates when the rule is used.
>>> If you use "all_done", it won't be skipped (but failure will also be
>>> counted as success) and when you use "none_failed" it will propagate the
>>> "failure" of B to C as well (but "skip" will not be propagated)..
>>>
>>> Generally speaking when you expect some task to be skippable you should
>>> design your DAG to account for it.
>>>
>>> You can see more about the rules in this really informative post by Marc
>>> Lamberti
>>> https://marclamberti.com/blog/airflow-trigger-rules-all-you-need-to-know/
>>>
>>> Whenever I can't remember how those rules work, I keep on coming back to
>>> the post.
>>>
>>>
>>>
>>>> James Coder
>>>> ------------------------------
>>>> *From:* Daniel Standish <da...@astronomer.io.INVALID>
>>>> *Sent:* Friday, February 4, 2022 1:41:12 AM
>>>> *To:* dev@airflow.apache.org <de...@airflow.apache.org>
>>>> *Subject:* Re: [DISCUSSION] Specify tasks to skip when triggering DAG
>>>>
>>>> That skip func had a typo (conf where it should have been context)...
>>>>
>>>> this is more likely to work:
>>>>
>>>> def skip_if_specified(context):
>>>> if not context:
>>>> return
>>>> dr = context.get('dag_run')
>>>> ti = context.get('task_instance')
>>>> if not (dr and ti):
>>>> return
>>>> conf = dr.conf
>>>> if not conf:
>>>> return
>>>> skip_list = conf.get('skip_list', [])
>>>> if ti.task_id in skip_list:
>>>> raise AirflowSkipException()
>>>>
>>>> Apologies for the spam.
>>>>
>>>
Re: [DISCUSSION] Specify tasks to skip when triggering DAG
Posted by Hongyi Wang <wh...@gmail.com>.
Hi Daniel,
Thank you for sharing the `pre_execute` idea. It is a smart idea, and I do
believe it would work. But to me, it is more as a walkaround instead of an
universal solution. To explain, I would love to discuss the tradeoffs.
[Pros]
The solution I proposed requires change in either Airflow Scheduler or
Executor. In comparison, the `pre_execute` solution doesn't require any
change, which avoids introducing additional risk/complexity.
[Cons]
1) Not scalable / Inconvenient
To make a task skippable, one needs to modify existing DAG (to set
`pre_execute`). It seems not difficult, but when your Airflow host a
thousand DAGs own by different teams/users, it can be challenging.
2) May cost additional time and resources
A task won’t be skipped until it reaches `pre_execute()`. It is okay when
using LocalExecutor/SequentialExecutor; however, when using
KubernetesExecutor, a k8s worker pod will always be created for “task to
skip”. Ideally, tasks are skipped during "scheduling".
3) May abuse `pre_execute`
Instead of asking developers to set `pre_execute=skip_if_specified` so the
task becomes skippable. Wouldn’t it be better if we make it (task is
skippable) an out-of-box feature, meanwhile `pre_execute` can serve its
original purpose?
Howie
On Fri, Feb 4, 2022 at 7:41 AM James Coder <jc...@gmail.com> wrote:
> Thanks Jarek, yeah, I wasn't thinking about the trigger rules, hazards of
> emailing before coffee.
>
> On Fri, Feb 4, 2022 at 10:32 AM Jarek Potiuk <ja...@potiuk.com> wrote:
>
>> Thanks Daniel, this something I have been thinking about for a while.
>>> One use case I have is for a dag only gets run on an ad hoc basis, for any
>>> combination of a subset of tasks (I recognize this is a rather non-standard
>>> use of airflow). One question though, if you specified skip_list=[“task b”]
>>> wouldn’t the scheduler skip task c as well?
>>>
>>
>> Depending on the triggering rule you choose.
>>
>> One of the effects of the default "all_success" rule is that the "skip"
>> state propagates when the rule is used.
>> If you use "all_done", it won't be skipped (but failure will also be
>> counted as success) and when you use "none_failed" it will propagate the
>> "failure" of B to C as well (but "skip" will not be propagated)..
>>
>> Generally speaking when you expect some task to be skippable you should
>> design your DAG to account for it.
>>
>> You can see more about the rules in this really informative post by Marc
>> Lamberti
>> https://marclamberti.com/blog/airflow-trigger-rules-all-you-need-to-know/
>>
>> Whenever I can't remember how those rules work, I keep on coming back to
>> the post.
>>
>>
>>
>>> James Coder
>>> ------------------------------
>>> *From:* Daniel Standish <da...@astronomer.io.INVALID>
>>> *Sent:* Friday, February 4, 2022 1:41:12 AM
>>> *To:* dev@airflow.apache.org <de...@airflow.apache.org>
>>> *Subject:* Re: [DISCUSSION] Specify tasks to skip when triggering DAG
>>>
>>> That skip func had a typo (conf where it should have been context)...
>>>
>>> this is more likely to work:
>>>
>>> def skip_if_specified(context):
>>> if not context:
>>> return
>>> dr = context.get('dag_run')
>>> ti = context.get('task_instance')
>>> if not (dr and ti):
>>> return
>>> conf = dr.conf
>>> if not conf:
>>> return
>>> skip_list = conf.get('skip_list', [])
>>> if ti.task_id in skip_list:
>>> raise AirflowSkipException()
>>>
>>> Apologies for the spam.
>>>
>>
Re: [DISCUSSION] Specify tasks to skip when triggering DAG
Posted by James Coder <jc...@gmail.com>.
Thanks Jarek, yeah, I wasn't thinking about the trigger rules, hazards of
emailing before coffee.
On Fri, Feb 4, 2022 at 10:32 AM Jarek Potiuk <ja...@potiuk.com> wrote:
> Thanks Daniel, this something I have been thinking about for a while. One
>> use case I have is for a dag only gets run on an ad hoc basis, for any
>> combination of a subset of tasks (I recognize this is a rather non-standard
>> use of airflow). One question though, if you specified skip_list=[“task b”]
>> wouldn’t the scheduler skip task c as well?
>>
>
> Depending on the triggering rule you choose.
>
> One of the effects of the default "all_success" rule is that the "skip"
> state propagates when the rule is used.
> If you use "all_done", it won't be skipped (but failure will also be
> counted as success) and when you use "none_failed" it will propagate the
> "failure" of B to C as well (but "skip" will not be propagated)..
>
> Generally speaking when you expect some task to be skippable you should
> design your DAG to account for it.
>
> You can see more about the rules in this really informative post by Marc
> Lamberti
> https://marclamberti.com/blog/airflow-trigger-rules-all-you-need-to-know/
>
> Whenever I can't remember how those rules work, I keep on coming back to
> the post.
>
>
>
>> James Coder
>> ------------------------------
>> *From:* Daniel Standish <da...@astronomer.io.INVALID>
>> *Sent:* Friday, February 4, 2022 1:41:12 AM
>> *To:* dev@airflow.apache.org <de...@airflow.apache.org>
>> *Subject:* Re: [DISCUSSION] Specify tasks to skip when triggering DAG
>>
>> That skip func had a typo (conf where it should have been context)...
>>
>> this is more likely to work:
>>
>> def skip_if_specified(context):
>> if not context:
>> return
>> dr = context.get('dag_run')
>> ti = context.get('task_instance')
>> if not (dr and ti):
>> return
>> conf = dr.conf
>> if not conf:
>> return
>> skip_list = conf.get('skip_list', [])
>> if ti.task_id in skip_list:
>> raise AirflowSkipException()
>>
>> Apologies for the spam.
>>
>
Re: [DISCUSSION] Specify tasks to skip when triggering DAG
Posted by Jarek Potiuk <ja...@potiuk.com>.
>
> Thanks Daniel, this something I have been thinking about for a while. One
> use case I have is for a dag only gets run on an ad hoc basis, for any
> combination of a subset of tasks (I recognize this is a rather non-standard
> use of airflow). One question though, if you specified skip_list=[“task b”]
> wouldn’t the scheduler skip task c as well?
>
Depending on the triggering rule you choose.
One of the effects of the default "all_success" rule is that the "skip"
state propagates when the rule is used.
If you use "all_done", it won't be skipped (but failure will also be
counted as success) and when you use "none_failed" it will propagate the
"failure" of B to C as well (but "skip" will not be propagated)..
Generally speaking when you expect some task to be skippable you should
design your DAG to account for it.
You can see more about the rules in this really informative post by Marc
Lamberti
https://marclamberti.com/blog/airflow-trigger-rules-all-you-need-to-know/
Whenever I can't remember how those rules work, I keep on coming back to
the post.
> James Coder
> ------------------------------
> *From:* Daniel Standish <da...@astronomer.io.INVALID>
> *Sent:* Friday, February 4, 2022 1:41:12 AM
> *To:* dev@airflow.apache.org <de...@airflow.apache.org>
> *Subject:* Re: [DISCUSSION] Specify tasks to skip when triggering DAG
>
> That skip func had a typo (conf where it should have been context)...
>
> this is more likely to work:
>
> def skip_if_specified(context):
> if not context:
> return
> dr = context.get('dag_run')
> ti = context.get('task_instance')
> if not (dr and ti):
> return
> conf = dr.conf
> if not conf:
> return
> skip_list = conf.get('skip_list', [])
> if ti.task_id in skip_list:
> raise AirflowSkipException()
>
> Apologies for the spam.
>
Re: [DISCUSSION] Specify tasks to skip when triggering DAG
Posted by James Coder <jc...@gmail.com>.
Thanks Daniel, this something I have been thinking about for a while. One use case I have is for a dag only gets run on an ad hoc basis, for any combination of a subset of tasks (I recognize this is a rather non-standard use of airflow). One question though, if you specified skip_list=[“task b”] wouldn’t the scheduler skip task c as well?
James Coder
________________________________
From: Daniel Standish <da...@astronomer.io.INVALID>
Sent: Friday, February 4, 2022 1:41:12 AM
To: dev@airflow.apache.org <de...@airflow.apache.org>
Subject: Re: [DISCUSSION] Specify tasks to skip when triggering DAG
That skip func had a typo (conf where it should have been context)...
this is more likely to work:
def skip_if_specified(context):
if not context:
return
dr = context.get('dag_run')
ti = context.get('task_instance')
if not (dr and ti):
return
conf = dr.conf
if not conf:
return
skip_list = conf.get('skip_list', [])
if ti.task_id in skip_list:
raise AirflowSkipException()
Apologies for the spam.
Re: [DISCUSSION] Specify tasks to skip when triggering DAG
Posted by Daniel Standish <da...@astronomer.io.INVALID>.
That skip func had a typo (conf where it should have been context)...
this is more likely to work:
def skip_if_specified(context):
if not context:
return
dr = context.get('dag_run')
ti = context.get('task_instance')
if not (dr and ti):
return
conf = dr.conf
if not conf:
return
skip_list = conf.get('skip_list', [])
if ti.task_id in skip_list:
raise AirflowSkipException()
Apologies for the spam.
Re: [DISCUSSION] Specify tasks to skip when triggering DAG
Posted by Daniel Standish <da...@astronomer.io.INVALID>.
Ok and here's an example you could try:
The callable:
def skip_if_specified(context):
dr = context.get('dag_run')
if not dr:
return
conf = dr.conf
if not conf:
return
skip_list = conf.get('skip_list')
if not skip_list:
return
ti = conf.get('task_instance')
if not ti:
return
if ti.task_id in skip_list:
raise AirflowSkipException()
How to use in a task:
op = BashOperator(
task_id='hello',
pre_execute=skip_if_specified,
bash_command='echo hello',
)
Or in all tasks in a dag:
dag = DAG(
task_id="my_dag",
default_args=dict(pre_execute=skip_if_specified),
...
)
And then, I'm pretty sure if you supply this conf to dag run it will skip
this task:
{'skip_list': ['hello']}
Re: [DISCUSSION] Specify tasks to skip when triggering DAG
Posted by Daniel Standish <da...@astronomer.io.INVALID>.
And if not with cluster policy, then you could pass such a "conditionally
skip" callable to all tasks in a dag with default_args
On Thu, Feb 3, 2022, 10:16 PM Daniel Standish <da...@astronomer.io>
wrote:
> It wouldn't necessarily need to involve the scheduler or the executor.
> You could add logic to pre_execute to read dag run conf and skip self is
> specified in the conf. In fact you could probably implement this currently
> with this approach using cluster policy, since we can supply pre_execute
> callables through param in base operator.
>
> On Thu, Feb 3, 2022, 10:07 PM Hongyi Wang <wh...@gmail.com> wrote:
>
>> Hello everyone,
>>
>> Just want to clarify a few points regarding "specify tasks to skip when
>> triggering DAG".
>>
>> First of all, the use case is not necessarily only backfill. We may want
>> to have this for manual triggers so users can do it easily in UI, rather
>> than using backfill CLI. This is similar to how Azkaban users skip a job
>> during execution (
>> https://azkaban.github.io/azkaban/docs/latest/#executing-flows).
>>
>> Second, we DO want to retain the dependencies and that is a “hard”
>> requirement (as is shown in the task-a ☐ -> task-b ☑ -> task-c ☐ example).
>>
>> Last but not least, the change does NOT necessarily have to be in
>> Scheduler, there is a chance it can be done in Executor instead (if that's
>> preferred).
>>
>> @Ash any objection to this proposal? If so, may you please share your
>> concern?
>>
>> Look forward to reaching a consensus soon.
>>
>> Howie
>>
>> On Mon, Jan 31, 2022 at 5:44 PM Hongyi Wang <wh...@gmail.com> wrote:
>>
>>> Hi Ash, thanks for the feedback.
>>>
>>> > You want it to behave as if the dag was specified as task-a -> task-c,
>>> right?
>>>
>>> Yes. As you mentioned, when we mark task-b as skipped, task-c's
>>> dependencies would be resolved and it would be eligible to run. BUT it does
>>> not necessarily mean task-a and task-c may run in parallel. The key is, if
>>> we only attempt to skip a task instance when it is eligible to run (e.g.
>>> task-b is eligible to run after task-a completes), we can still retain
>>> dependencies between the rest of tasks.
>>>
>>> > I'm not sold that this should belong in Airflow scheduler --
>>> re-running DagRuns on an ad-hoc basis is more aligned with `airflow
>>> backfill`.
>>>
>>> Are you suggesting "skipping tasks when triggering DAG" should be an
>>> option specific to `airflow backfill`? (please correct me if I
>>> misunderstand) To me, the proposal is more aligned with `airflow trigger`
>>> than `airflow backfill`, but I hear you. The reason why I believe
>>> `scheduler_job.py` is the best place to skip tasks is because it allows
>>> Airflow to retain dependencies between non-skipped tasks (as I explained
>>> above). I understand changes to the Airflow scheduler may impact all users.
>>> I won't deny there is a risk. But considering the code change is
>>> comparatively small and straightforward, I am not too worried once it has
>>> been reviewed and tested.
>>>
>>> Hope I answered your question. I am happy to discuss more.
>>>
>>> Howie
>>>
>>> On Mon, Jan 31, 2022 at 4:03 AM Ash Berlin-Taylor <as...@apache.org>
>>> wrote:
>>>
>>>> > To illustrate the use case, I am going to use this example below.
>>>> > task-a ☐ -> task-b ☑ -> task-c ☐
>>>>
>>>> So in this example, are you aware and intending that both task-a and
>>>> task-c run straight away? Because by skipping task-b, task-c's dependencies
>>>> would be resolved and it would be eligible to run.
>>>>
>>>> From what you've describe I don't think that is actually what you want,
>>>> but that you want it to behave as if the dag was specified as task-a ->
>>>> task-c, right?
>>>>
>>>> Honestly though: I'm not sold that this should belong in Airflow
>>>> scheduler -- re-running DagRuns on an ad-hoc basis is more aligned with
>>>> `airflow backfill`.
>>>>
>>>> -ash
>>>>
>>>> On Sun, Jan 30 2022 at 11:21:26 -0800, Hongyi Wang <wh...@gmail.com>
>>>> wrote:
>>>>
>>>> Hi Elad, thank you for your feedback. To answer your question, besides
>>>> debugging, another common use case is re-running existing DAGs in an ad-hoc
>>>> manner.
>>>>
>>>> For example, as an Airflow user, I sometimes want to trigger an ad-hoc
>>>> DAG run. In this run, I want to skip one/more tasks, so the dag run can
>>>> yield a different result, or simply complete sooner. As I mentioned in my
>>>> previous email, there are other ways to achieve the same goal. But IMHO,
>>>> neither of them are easy & flexible enough for an ad-hoc use case.
>>>>
>>>> Does that sound like a reasonable use case? What do you think is the
>>>> best approach to solve it? I am happy to discuss more with you.
>>>>
>>>> On Sun, Jan 30, 2022 at 4:45 AM Elad Kalif <el...@apache.org> wrote:
>>>>
>>>>> Can you describe a use case for the requested feature other than
>>>>> debugging? This doesn't feel like the right approach to test a specific
>>>>> task in a pipeline.
>>>>>
>>>>> On Fri, Jan 28, 2022 at 11:44 PM Alex Begg <al...@gmail.com>
>>>>> wrote:
>>>>>
>>>>>> Actually, sorry, you can scratch out some of what I just said, I
>>>>>> thought you were talking about clearing states, you are instead referring
>>>>>> to triggering a DAG run. That does kind of make sense to have a way to
>>>>>> trigger a DAG run but only run specific tasks.
>>>>>>
>>>>>> On Fri, Jan 28, 2022 at 1:41 PM Alex Begg <al...@gmail.com>
>>>>>> wrote:
>>>>>>
>>>>>>> I believe this is currently possible by just unselecting
>>>>>>> “downstream” before you click “Clear” in the UI. It should only clear the
>>>>>>> one middle task and not the downstream task(s).
>>>>>>>
>>>>>>> I would prefer to not have a more detailed UI to allow to skip (or i
>>>>>>> want to say “bypass” as “skip” is itself a task state) specific downstream
>>>>>>> tasks as it might signal to users that it is ideal to specify tasks to
>>>>>>> bypass when in reality it is only something that should be done on occasion
>>>>>>> for experiment or troubleshooting as you mention, not a common occurrence.
>>>>>>>
>>>>>>> What I can agree to though is the list of buttons on the dialog
>>>>>>> window to change state of a task is a bit cluttered looking. There probably
>>>>>>> can be a better UI/UX for that, but I do think being able to check/uncheck
>>>>>>> downstream task is a way to go, that seems like it will be just as
>>>>>>> cluttered.
>>>>>>>
>>>>>>> Alex Begg
>>>>>>>
>>>>>>> On Fri, Jan 28, 2022 at 11:46 AM Hongyi Wang <wh...@gmail.com>
>>>>>>> wrote:
>>>>>>>
>>>>>>>> Hello everyone,
>>>>>>>>
>>>>>>>> I'd like to propose a new feature in Airflow -- allow users to specify tasks to skip when trigger DAG run.
>>>>>>>>
>>>>>>>> From our own experience, this feature can be very useful when doing experiments, troubleshooting or re-running existing DAGs. And I believe it can benefit many Airflow users.
>>>>>>>>
>>>>>>>> To illustrate the use case, I am going to use this example below.
>>>>>>>> task-a ☐ -> task-b ☑ -> task-c ☐
>>>>>>>>
>>>>>>>> Suppose we have a DAG containing 3 tasks. To troubleshoot "task-a" and "task-c", I want to trigger a manual DAG run and skip "task-b" (so I can save time & resource & focus on other two tasks). To do so, today I have two options:
>>>>>>>>
>>>>>>>> Option 1: Trigger DAG, then manually mark "task-b" as `SUCCESS`
>>>>>>>> Option 2: Remove "task-b" from my DAG, then trigger DAG
>>>>>>>>
>>>>>>>> Neither of the options are great. Option 1 can be troublesome when DAG is large, and there are multiple tasks I want to skip. Option 2 requires change in the DAG file, which is not convenient for just troubleshooting.
>>>>>>>>
>>>>>>>> Therefore, I would love to discuss how we can provide an easy way for users to skip tasks when triggering DAG.
>>>>>>>>
>>>>>>>> Things to consider are:
>>>>>>>> 1) We should allow user to specify all tasks to skip at once when trigger DAG
>>>>>>>> 2) We should retain the dependencies between non-skip tasks (in above example, "task-c" won't start until "task-a" completes even if we skipped "task-b")
>>>>>>>> 3) We should mark skipped task as `SKIPPED` instead of `SUCCESS` to make it more intuitive
>>>>>>>> 4) The implementation should be easy, clean and low risk
>>>>>>>>
>>>>>>>> Here is my proposed solution (tested locally):
>>>>>>>> Today, Airflow allow user to pass a JSON to the Dagrun as {{dag_run.conf}} when triggering DAG. The idea is, before queuing task instances that satisfies dependences, `scheduler_job.py` (after we make some change) will filter task instances to skip based on `dag_run.conf` user passes in (e.g. {"skip_tasks": ["task-b"]}), then mark them as SKIPPED.
>>>>>>>>
>>>>>>>> Things I would love to discuss:
>>>>>>>> - What do you think about this feature?
>>>>>>>> - What do you think about the proposed solution?
>>>>>>>> - Did I miss anything that you want to discuss?
>>>>>>>> - Is it necessary to introduce a new state (e.g. MANUAL_SKIPPED) to differentiate SKIPPED?
>>>>>>>>
>>>>>>>> Howie
>>>>>>>>
>>>>>>>>
Re: [DISCUSSION] Specify tasks to skip when triggering DAG
Posted by Daniel Standish <da...@astronomer.io.INVALID>.
It wouldn't necessarily need to involve the scheduler or the executor. You
could add logic to pre_execute to read dag run conf and skip self is
specified in the conf. In fact you could probably implement this currently
with this approach using cluster policy, since we can supply pre_execute
callables through param in base operator.
On Thu, Feb 3, 2022, 10:07 PM Hongyi Wang <wh...@gmail.com> wrote:
> Hello everyone,
>
> Just want to clarify a few points regarding "specify tasks to skip when
> triggering DAG".
>
> First of all, the use case is not necessarily only backfill. We may want
> to have this for manual triggers so users can do it easily in UI, rather
> than using backfill CLI. This is similar to how Azkaban users skip a job
> during execution (
> https://azkaban.github.io/azkaban/docs/latest/#executing-flows).
>
> Second, we DO want to retain the dependencies and that is a “hard”
> requirement (as is shown in the task-a ☐ -> task-b ☑ -> task-c ☐ example).
>
> Last but not least, the change does NOT necessarily have to be in
> Scheduler, there is a chance it can be done in Executor instead (if that's
> preferred).
>
> @Ash any objection to this proposal? If so, may you please share your
> concern?
>
> Look forward to reaching a consensus soon.
>
> Howie
>
> On Mon, Jan 31, 2022 at 5:44 PM Hongyi Wang <wh...@gmail.com> wrote:
>
>> Hi Ash, thanks for the feedback.
>>
>> > You want it to behave as if the dag was specified as task-a -> task-c,
>> right?
>>
>> Yes. As you mentioned, when we mark task-b as skipped, task-c's
>> dependencies would be resolved and it would be eligible to run. BUT it does
>> not necessarily mean task-a and task-c may run in parallel. The key is, if
>> we only attempt to skip a task instance when it is eligible to run (e.g.
>> task-b is eligible to run after task-a completes), we can still retain
>> dependencies between the rest of tasks.
>>
>> > I'm not sold that this should belong in Airflow scheduler --
>> re-running DagRuns on an ad-hoc basis is more aligned with `airflow
>> backfill`.
>>
>> Are you suggesting "skipping tasks when triggering DAG" should be an
>> option specific to `airflow backfill`? (please correct me if I
>> misunderstand) To me, the proposal is more aligned with `airflow trigger`
>> than `airflow backfill`, but I hear you. The reason why I believe
>> `scheduler_job.py` is the best place to skip tasks is because it allows
>> Airflow to retain dependencies between non-skipped tasks (as I explained
>> above). I understand changes to the Airflow scheduler may impact all users.
>> I won't deny there is a risk. But considering the code change is
>> comparatively small and straightforward, I am not too worried once it has
>> been reviewed and tested.
>>
>> Hope I answered your question. I am happy to discuss more.
>>
>> Howie
>>
>> On Mon, Jan 31, 2022 at 4:03 AM Ash Berlin-Taylor <as...@apache.org> wrote:
>>
>>> > To illustrate the use case, I am going to use this example below.
>>> > task-a ☐ -> task-b ☑ -> task-c ☐
>>>
>>> So in this example, are you aware and intending that both task-a and
>>> task-c run straight away? Because by skipping task-b, task-c's dependencies
>>> would be resolved and it would be eligible to run.
>>>
>>> From what you've describe I don't think that is actually what you want,
>>> but that you want it to behave as if the dag was specified as task-a ->
>>> task-c, right?
>>>
>>> Honestly though: I'm not sold that this should belong in Airflow
>>> scheduler -- re-running DagRuns on an ad-hoc basis is more aligned with
>>> `airflow backfill`.
>>>
>>> -ash
>>>
>>> On Sun, Jan 30 2022 at 11:21:26 -0800, Hongyi Wang <wh...@gmail.com>
>>> wrote:
>>>
>>> Hi Elad, thank you for your feedback. To answer your question, besides
>>> debugging, another common use case is re-running existing DAGs in an ad-hoc
>>> manner.
>>>
>>> For example, as an Airflow user, I sometimes want to trigger an ad-hoc
>>> DAG run. In this run, I want to skip one/more tasks, so the dag run can
>>> yield a different result, or simply complete sooner. As I mentioned in my
>>> previous email, there are other ways to achieve the same goal. But IMHO,
>>> neither of them are easy & flexible enough for an ad-hoc use case.
>>>
>>> Does that sound like a reasonable use case? What do you think is the
>>> best approach to solve it? I am happy to discuss more with you.
>>>
>>> On Sun, Jan 30, 2022 at 4:45 AM Elad Kalif <el...@apache.org> wrote:
>>>
>>>> Can you describe a use case for the requested feature other than
>>>> debugging? This doesn't feel like the right approach to test a specific
>>>> task in a pipeline.
>>>>
>>>> On Fri, Jan 28, 2022 at 11:44 PM Alex Begg <al...@gmail.com> wrote:
>>>>
>>>>> Actually, sorry, you can scratch out some of what I just said, I
>>>>> thought you were talking about clearing states, you are instead referring
>>>>> to triggering a DAG run. That does kind of make sense to have a way to
>>>>> trigger a DAG run but only run specific tasks.
>>>>>
>>>>> On Fri, Jan 28, 2022 at 1:41 PM Alex Begg <al...@gmail.com> wrote:
>>>>>
>>>>>> I believe this is currently possible by just unselecting “downstream”
>>>>>> before you click “Clear” in the UI. It should only clear the one middle
>>>>>> task and not the downstream task(s).
>>>>>>
>>>>>> I would prefer to not have a more detailed UI to allow to skip (or i
>>>>>> want to say “bypass” as “skip” is itself a task state) specific downstream
>>>>>> tasks as it might signal to users that it is ideal to specify tasks to
>>>>>> bypass when in reality it is only something that should be done on occasion
>>>>>> for experiment or troubleshooting as you mention, not a common occurrence.
>>>>>>
>>>>>> What I can agree to though is the list of buttons on the dialog
>>>>>> window to change state of a task is a bit cluttered looking. There probably
>>>>>> can be a better UI/UX for that, but I do think being able to check/uncheck
>>>>>> downstream task is a way to go, that seems like it will be just as
>>>>>> cluttered.
>>>>>>
>>>>>> Alex Begg
>>>>>>
>>>>>> On Fri, Jan 28, 2022 at 11:46 AM Hongyi Wang <wh...@gmail.com>
>>>>>> wrote:
>>>>>>
>>>>>>> Hello everyone,
>>>>>>>
>>>>>>> I'd like to propose a new feature in Airflow -- allow users to specify tasks to skip when trigger DAG run.
>>>>>>>
>>>>>>> From our own experience, this feature can be very useful when doing experiments, troubleshooting or re-running existing DAGs. And I believe it can benefit many Airflow users.
>>>>>>>
>>>>>>> To illustrate the use case, I am going to use this example below.
>>>>>>> task-a ☐ -> task-b ☑ -> task-c ☐
>>>>>>>
>>>>>>> Suppose we have a DAG containing 3 tasks. To troubleshoot "task-a" and "task-c", I want to trigger a manual DAG run and skip "task-b" (so I can save time & resource & focus on other two tasks). To do so, today I have two options:
>>>>>>>
>>>>>>> Option 1: Trigger DAG, then manually mark "task-b" as `SUCCESS`
>>>>>>> Option 2: Remove "task-b" from my DAG, then trigger DAG
>>>>>>>
>>>>>>> Neither of the options are great. Option 1 can be troublesome when DAG is large, and there are multiple tasks I want to skip. Option 2 requires change in the DAG file, which is not convenient for just troubleshooting.
>>>>>>>
>>>>>>> Therefore, I would love to discuss how we can provide an easy way for users to skip tasks when triggering DAG.
>>>>>>>
>>>>>>> Things to consider are:
>>>>>>> 1) We should allow user to specify all tasks to skip at once when trigger DAG
>>>>>>> 2) We should retain the dependencies between non-skip tasks (in above example, "task-c" won't start until "task-a" completes even if we skipped "task-b")
>>>>>>> 3) We should mark skipped task as `SKIPPED` instead of `SUCCESS` to make it more intuitive
>>>>>>> 4) The implementation should be easy, clean and low risk
>>>>>>>
>>>>>>> Here is my proposed solution (tested locally):
>>>>>>> Today, Airflow allow user to pass a JSON to the Dagrun as {{dag_run.conf}} when triggering DAG. The idea is, before queuing task instances that satisfies dependences, `scheduler_job.py` (after we make some change) will filter task instances to skip based on `dag_run.conf` user passes in (e.g. {"skip_tasks": ["task-b"]}), then mark them as SKIPPED.
>>>>>>>
>>>>>>> Things I would love to discuss:
>>>>>>> - What do you think about this feature?
>>>>>>> - What do you think about the proposed solution?
>>>>>>> - Did I miss anything that you want to discuss?
>>>>>>> - Is it necessary to introduce a new state (e.g. MANUAL_SKIPPED) to differentiate SKIPPED?
>>>>>>>
>>>>>>> Howie
>>>>>>>
>>>>>>>
Re: [DISCUSSION] Specify tasks to skip when triggering DAG
Posted by Hongyi Wang <wh...@gmail.com>.
Hello everyone,
Just want to clarify a few points regarding "specify tasks to skip when
triggering DAG".
First of all, the use case is not necessarily only backfill. We may want to
have this for manual triggers so users can do it easily in UI, rather than
using backfill CLI. This is similar to how Azkaban users skip a job during
execution (https://azkaban.github.io/azkaban/docs/latest/#executing-flows).
Second, we DO want to retain the dependencies and that is a “hard”
requirement (as is shown in the task-a ☐ -> task-b ☑ -> task-c ☐ example).
Last but not least, the change does NOT necessarily have to be in
Scheduler, there is a chance it can be done in Executor instead (if that's
preferred).
@Ash any objection to this proposal? If so, may you please share your
concern?
Look forward to reaching a consensus soon.
Howie
On Mon, Jan 31, 2022 at 5:44 PM Hongyi Wang <wh...@gmail.com> wrote:
> Hi Ash, thanks for the feedback.
>
> > You want it to behave as if the dag was specified as task-a -> task-c,
> right?
>
> Yes. As you mentioned, when we mark task-b as skipped, task-c's
> dependencies would be resolved and it would be eligible to run. BUT it does
> not necessarily mean task-a and task-c may run in parallel. The key is, if
> we only attempt to skip a task instance when it is eligible to run (e.g.
> task-b is eligible to run after task-a completes), we can still retain
> dependencies between the rest of tasks.
>
> > I'm not sold that this should belong in Airflow scheduler -- re-running
> DagRuns on an ad-hoc basis is more aligned with `airflow backfill`.
>
> Are you suggesting "skipping tasks when triggering DAG" should be an
> option specific to `airflow backfill`? (please correct me if I
> misunderstand) To me, the proposal is more aligned with `airflow trigger`
> than `airflow backfill`, but I hear you. The reason why I believe
> `scheduler_job.py` is the best place to skip tasks is because it allows
> Airflow to retain dependencies between non-skipped tasks (as I explained
> above). I understand changes to the Airflow scheduler may impact all users.
> I won't deny there is a risk. But considering the code change is
> comparatively small and straightforward, I am not too worried once it has
> been reviewed and tested.
>
> Hope I answered your question. I am happy to discuss more.
>
> Howie
>
> On Mon, Jan 31, 2022 at 4:03 AM Ash Berlin-Taylor <as...@apache.org> wrote:
>
>> > To illustrate the use case, I am going to use this example below.
>> > task-a ☐ -> task-b ☑ -> task-c ☐
>>
>> So in this example, are you aware and intending that both task-a and
>> task-c run straight away? Because by skipping task-b, task-c's dependencies
>> would be resolved and it would be eligible to run.
>>
>> From what you've describe I don't think that is actually what you want,
>> but that you want it to behave as if the dag was specified as task-a ->
>> task-c, right?
>>
>> Honestly though: I'm not sold that this should belong in Airflow
>> scheduler -- re-running DagRuns on an ad-hoc basis is more aligned with
>> `airflow backfill`.
>>
>> -ash
>>
>> On Sun, Jan 30 2022 at 11:21:26 -0800, Hongyi Wang <wh...@gmail.com>
>> wrote:
>>
>> Hi Elad, thank you for your feedback. To answer your question, besides
>> debugging, another common use case is re-running existing DAGs in an ad-hoc
>> manner.
>>
>> For example, as an Airflow user, I sometimes want to trigger an ad-hoc
>> DAG run. In this run, I want to skip one/more tasks, so the dag run can
>> yield a different result, or simply complete sooner. As I mentioned in my
>> previous email, there are other ways to achieve the same goal. But IMHO,
>> neither of them are easy & flexible enough for an ad-hoc use case.
>>
>> Does that sound like a reasonable use case? What do you think is the best
>> approach to solve it? I am happy to discuss more with you.
>>
>> On Sun, Jan 30, 2022 at 4:45 AM Elad Kalif <el...@apache.org> wrote:
>>
>>> Can you describe a use case for the requested feature other than
>>> debugging? This doesn't feel like the right approach to test a specific
>>> task in a pipeline.
>>>
>>> On Fri, Jan 28, 2022 at 11:44 PM Alex Begg <al...@gmail.com> wrote:
>>>
>>>> Actually, sorry, you can scratch out some of what I just said, I
>>>> thought you were talking about clearing states, you are instead referring
>>>> to triggering a DAG run. That does kind of make sense to have a way to
>>>> trigger a DAG run but only run specific tasks.
>>>>
>>>> On Fri, Jan 28, 2022 at 1:41 PM Alex Begg <al...@gmail.com> wrote:
>>>>
>>>>> I believe this is currently possible by just unselecting “downstream”
>>>>> before you click “Clear” in the UI. It should only clear the one middle
>>>>> task and not the downstream task(s).
>>>>>
>>>>> I would prefer to not have a more detailed UI to allow to skip (or i
>>>>> want to say “bypass” as “skip” is itself a task state) specific downstream
>>>>> tasks as it might signal to users that it is ideal to specify tasks to
>>>>> bypass when in reality it is only something that should be done on occasion
>>>>> for experiment or troubleshooting as you mention, not a common occurrence.
>>>>>
>>>>> What I can agree to though is the list of buttons on the dialog window
>>>>> to change state of a task is a bit cluttered looking. There probably can be
>>>>> a better UI/UX for that, but I do think being able to check/uncheck
>>>>> downstream task is a way to go, that seems like it will be just as
>>>>> cluttered.
>>>>>
>>>>> Alex Begg
>>>>>
>>>>> On Fri, Jan 28, 2022 at 11:46 AM Hongyi Wang <wh...@gmail.com>
>>>>> wrote:
>>>>>
>>>>>> Hello everyone,
>>>>>>
>>>>>> I'd like to propose a new feature in Airflow -- allow users to specify tasks to skip when trigger DAG run.
>>>>>>
>>>>>> From our own experience, this feature can be very useful when doing experiments, troubleshooting or re-running existing DAGs. And I believe it can benefit many Airflow users.
>>>>>>
>>>>>> To illustrate the use case, I am going to use this example below.
>>>>>> task-a ☐ -> task-b ☑ -> task-c ☐
>>>>>>
>>>>>> Suppose we have a DAG containing 3 tasks. To troubleshoot "task-a" and "task-c", I want to trigger a manual DAG run and skip "task-b" (so I can save time & resource & focus on other two tasks). To do so, today I have two options:
>>>>>>
>>>>>> Option 1: Trigger DAG, then manually mark "task-b" as `SUCCESS`
>>>>>> Option 2: Remove "task-b" from my DAG, then trigger DAG
>>>>>>
>>>>>> Neither of the options are great. Option 1 can be troublesome when DAG is large, and there are multiple tasks I want to skip. Option 2 requires change in the DAG file, which is not convenient for just troubleshooting.
>>>>>>
>>>>>> Therefore, I would love to discuss how we can provide an easy way for users to skip tasks when triggering DAG.
>>>>>>
>>>>>> Things to consider are:
>>>>>> 1) We should allow user to specify all tasks to skip at once when trigger DAG
>>>>>> 2) We should retain the dependencies between non-skip tasks (in above example, "task-c" won't start until "task-a" completes even if we skipped "task-b")
>>>>>> 3) We should mark skipped task as `SKIPPED` instead of `SUCCESS` to make it more intuitive
>>>>>> 4) The implementation should be easy, clean and low risk
>>>>>>
>>>>>> Here is my proposed solution (tested locally):
>>>>>> Today, Airflow allow user to pass a JSON to the Dagrun as {{dag_run.conf}} when triggering DAG. The idea is, before queuing task instances that satisfies dependences, `scheduler_job.py` (after we make some change) will filter task instances to skip based on `dag_run.conf` user passes in (e.g. {"skip_tasks": ["task-b"]}), then mark them as SKIPPED.
>>>>>>
>>>>>> Things I would love to discuss:
>>>>>> - What do you think about this feature?
>>>>>> - What do you think about the proposed solution?
>>>>>> - Did I miss anything that you want to discuss?
>>>>>> - Is it necessary to introduce a new state (e.g. MANUAL_SKIPPED) to differentiate SKIPPED?
>>>>>>
>>>>>> Howie
>>>>>>
>>>>>>
Re: [DISCUSSION] Specify tasks to skip when triggering DAG
Posted by Hongyi Wang <wh...@gmail.com>.
Hi Ash, thanks for the feedback.
> You want it to behave as if the dag was specified as task-a -> task-c,
right?
Yes. As you mentioned, when we mark task-b as skipped, task-c's
dependencies would be resolved and it would be eligible to run. BUT it does
not necessarily mean task-a and task-c may run in parallel. The key is, if
we only attempt to skip a task instance when it is eligible to run (e.g.
task-b is eligible to run after task-a completes), we can still retain
dependencies between the rest of tasks.
> I'm not sold that this should belong in Airflow scheduler -- re-running
DagRuns on an ad-hoc basis is more aligned with `airflow backfill`.
Are you suggesting "skipping tasks when triggering DAG" should be an option
specific to `airflow backfill`? (please correct me if I misunderstand) To
me, the proposal is more aligned with `airflow trigger` than `airflow
backfill`, but I hear you. The reason why I believe `scheduler_job.py` is
the best place to skip tasks is because it allows Airflow to retain
dependencies between non-skipped tasks (as I explained above). I understand
changes to the Airflow scheduler may impact all users. I won't deny there
is a risk. But considering the code change is comparatively small and
straightforward, I am not too worried once it has been reviewed and tested.
Hope I answered your question. I am happy to discuss more.
Howie
On Mon, Jan 31, 2022 at 4:03 AM Ash Berlin-Taylor <as...@apache.org> wrote:
> > To illustrate the use case, I am going to use this example below.
> > task-a ☐ -> task-b ☑ -> task-c ☐
>
> So in this example, are you aware and intending that both task-a and
> task-c run straight away? Because by skipping task-b, task-c's dependencies
> would be resolved and it would be eligible to run.
>
> From what you've describe I don't think that is actually what you want,
> but that you want it to behave as if the dag was specified as task-a ->
> task-c, right?
>
> Honestly though: I'm not sold that this should belong in Airflow scheduler
> -- re-running DagRuns on an ad-hoc basis is more aligned with `airflow
> backfill`.
>
> -ash
>
> On Sun, Jan 30 2022 at 11:21:26 -0800, Hongyi Wang <wh...@gmail.com>
> wrote:
>
> Hi Elad, thank you for your feedback. To answer your question, besides
> debugging, another common use case is re-running existing DAGs in an ad-hoc
> manner.
>
> For example, as an Airflow user, I sometimes want to trigger an ad-hoc DAG
> run. In this run, I want to skip one/more tasks, so the dag run can yield a
> different result, or simply complete sooner. As I mentioned in my previous
> email, there are other ways to achieve the same goal. But IMHO, neither of
> them are easy & flexible enough for an ad-hoc use case.
>
> Does that sound like a reasonable use case? What do you think is the best
> approach to solve it? I am happy to discuss more with you.
>
> On Sun, Jan 30, 2022 at 4:45 AM Elad Kalif <el...@apache.org> wrote:
>
>> Can you describe a use case for the requested feature other than
>> debugging? This doesn't feel like the right approach to test a specific
>> task in a pipeline.
>>
>> On Fri, Jan 28, 2022 at 11:44 PM Alex Begg <al...@gmail.com> wrote:
>>
>>> Actually, sorry, you can scratch out some of what I just said, I thought
>>> you were talking about clearing states, you are instead referring to
>>> triggering a DAG run. That does kind of make sense to have a way to trigger
>>> a DAG run but only run specific tasks.
>>>
>>> On Fri, Jan 28, 2022 at 1:41 PM Alex Begg <al...@gmail.com> wrote:
>>>
>>>> I believe this is currently possible by just unselecting “downstream”
>>>> before you click “Clear” in the UI. It should only clear the one middle
>>>> task and not the downstream task(s).
>>>>
>>>> I would prefer to not have a more detailed UI to allow to skip (or i
>>>> want to say “bypass” as “skip” is itself a task state) specific downstream
>>>> tasks as it might signal to users that it is ideal to specify tasks to
>>>> bypass when in reality it is only something that should be done on occasion
>>>> for experiment or troubleshooting as you mention, not a common occurrence.
>>>>
>>>> What I can agree to though is the list of buttons on the dialog window
>>>> to change state of a task is a bit cluttered looking. There probably can be
>>>> a better UI/UX for that, but I do think being able to check/uncheck
>>>> downstream task is a way to go, that seems like it will be just as
>>>> cluttered.
>>>>
>>>> Alex Begg
>>>>
>>>> On Fri, Jan 28, 2022 at 11:46 AM Hongyi Wang <wh...@gmail.com>
>>>> wrote:
>>>>
>>>>> Hello everyone,
>>>>>
>>>>> I'd like to propose a new feature in Airflow -- allow users to specify tasks to skip when trigger DAG run.
>>>>>
>>>>> From our own experience, this feature can be very useful when doing experiments, troubleshooting or re-running existing DAGs. And I believe it can benefit many Airflow users.
>>>>>
>>>>> To illustrate the use case, I am going to use this example below.
>>>>> task-a ☐ -> task-b ☑ -> task-c ☐
>>>>>
>>>>> Suppose we have a DAG containing 3 tasks. To troubleshoot "task-a" and "task-c", I want to trigger a manual DAG run and skip "task-b" (so I can save time & resource & focus on other two tasks). To do so, today I have two options:
>>>>>
>>>>> Option 1: Trigger DAG, then manually mark "task-b" as `SUCCESS`
>>>>> Option 2: Remove "task-b" from my DAG, then trigger DAG
>>>>>
>>>>> Neither of the options are great. Option 1 can be troublesome when DAG is large, and there are multiple tasks I want to skip. Option 2 requires change in the DAG file, which is not convenient for just troubleshooting.
>>>>>
>>>>> Therefore, I would love to discuss how we can provide an easy way for users to skip tasks when triggering DAG.
>>>>>
>>>>> Things to consider are:
>>>>> 1) We should allow user to specify all tasks to skip at once when trigger DAG
>>>>> 2) We should retain the dependencies between non-skip tasks (in above example, "task-c" won't start until "task-a" completes even if we skipped "task-b")
>>>>> 3) We should mark skipped task as `SKIPPED` instead of `SUCCESS` to make it more intuitive
>>>>> 4) The implementation should be easy, clean and low risk
>>>>>
>>>>> Here is my proposed solution (tested locally):
>>>>> Today, Airflow allow user to pass a JSON to the Dagrun as {{dag_run.conf}} when triggering DAG. The idea is, before queuing task instances that satisfies dependences, `scheduler_job.py` (after we make some change) will filter task instances to skip based on `dag_run.conf` user passes in (e.g. {"skip_tasks": ["task-b"]}), then mark them as SKIPPED.
>>>>>
>>>>> Things I would love to discuss:
>>>>> - What do you think about this feature?
>>>>> - What do you think about the proposed solution?
>>>>> - Did I miss anything that you want to discuss?
>>>>> - Is it necessary to introduce a new state (e.g. MANUAL_SKIPPED) to differentiate SKIPPED?
>>>>>
>>>>> Howie
>>>>>
>>>>>
Re: [DISCUSSION] Specify tasks to skip when triggering DAG
Posted by Ash Berlin-Taylor <as...@apache.org>.
> To illustrate the use case, I am going to use this example below.
> task-a ☐ -> task-b ☑ -> task-c ☐
So in this example, are you aware and intending that both task-a and
task-c run straight away? Because by skipping task-b, task-c's
dependencies would be resolved and it would be eligible to run.
From what you've describe I don't think that is actually what you
want, but that you want it to behave as if the dag was specified as
task-a -> task-c, right?
Honestly though: I'm not sold that this should belong in Airflow
scheduler -- re-running DagRuns on an ad-hoc basis is more aligned with
`airflow backfill`.
-ash
On Sun, Jan 30 2022 at 11:21:26 -0800, Hongyi Wang <wh...@gmail.com>
wrote:
> Hi Elad, thank you for your feedback. To answer your question,
> besides debugging, another common use case is re-running existing
> DAGs in an ad-hoc manner.
>
> For example, as an Airflow user, I sometimes want to trigger an
> ad-hoc DAG run. In this run, I want to skip one/more tasks, so the
> dag run can yield a different result, or simply complete sooner. As I
> mentioned in my previous email, there are other ways to achieve the
> same goal. But IMHO, neither of them are easy & flexible enough for
> an ad-hoc use case.
>
> Does that sound like a reasonable use case? What do you think is the
> best approach to solve it? I am happy to discuss more with you.
>
> On Sun, Jan 30, 2022 at 4:45 AM Elad Kalif <eladkal@apache.org
> <ma...@apache.org>> wrote:
>> Can you describe a use case for the requested feature other than
>> debugging? This doesn't feel like the right approach to test a
>> specific task in a pipeline.
>>
>> On Fri, Jan 28, 2022 at 11:44 PM Alex Begg <alex.begg@gmail.com
>> <ma...@gmail.com>> wrote:
>>> Actually, sorry, you can scratch out some of what I just said, I
>>> thought you were talking about clearing states, you are instead
>>> referring to triggering a DAG run. That does kind of make sense to
>>> have a way to trigger a DAG run but only run specific tasks.
>>>
>>> On Fri, Jan 28, 2022 at 1:41 PM Alex Begg <alex.begg@gmail.com
>>> <ma...@gmail.com>> wrote:
>>>> I believe this is currently possible by just unselecting
>>>> “downstream” before you click “Clear” in the UI. It should
>>>> only clear the one middle task and not the downstream task(s).
>>>>
>>>> I would prefer to not have a more detailed UI to allow to skip (or
>>>> i want to say “bypass” as “skip” is itself a task state)
>>>> specific downstream tasks as it might signal to users that it is
>>>> ideal to specify tasks to bypass when in reality it is only
>>>> something that should be done on occasion for experiment or
>>>> troubleshooting as you mention, not a common occurrence.
>>>>
>>>> What I can agree to though is the list of buttons on the dialog
>>>> window to change state of a task is a bit cluttered looking. There
>>>> probably can be a better UI/UX for that, but I do think being able
>>>> to check/uncheck downstream task is a way to go, that seems like
>>>> it will be just as cluttered.
>>>>
>>>> Alex Begg
>>>>
>>>> On Fri, Jan 28, 2022 at 11:46 AM Hongyi Wang <whynick1@gmail.com
>>>> <ma...@gmail.com>> wrote:
>>>>> Hello everyone,
>>>>>
>>>>> I'd like to propose a new feature in Airflow -- allow users to
>>>>> specify tasks to skip when trigger DAG run.
>>>>> From our own experience, this feature can be very useful when
>>>>> doing experiments, troubleshooting or re-running existing DAGs.
>>>>> And I believe it can benefit many Airflow users.
>>>>> To illustrate the use case, I am going to use this example below.
>>>>> task-a ☐ -> task-b ☑ -> task-c ☐
>>>>>
>>>>> Suppose we have a DAG containing 3 tasks. To troubleshoot
>>>>> "task-a" and "task-c", I want to trigger a manual DAG run and
>>>>> skip "task-b" (so I can save time & resource & focus on other two
>>>>> tasks). To do so, today I have two options:
>>>>>
>>>>> Option 1: Trigger DAG, then manually mark "task-b" as `SUCCESS`
>>>>> Option 2: Remove "task-b" from my DAG, then trigger DAG
>>>>>
>>>>> Neither of the options are great. Option 1 can be troublesome
>>>>> when DAG is large, and there are multiple tasks I want to skip.
>>>>> Option 2 requires change in the DAG file, which is not convenient
>>>>> for just troubleshooting.
>>>>>
>>>>> Therefore, I would love to discuss how we can provide an easy way
>>>>> for users to skip tasks when triggering DAG.
>>>>>
>>>>> Things to consider are:
>>>>> 1) We should allow user to specify all tasks to skip at once when
>>>>> trigger DAG
>>>>> 2) We should retain the dependencies between non-skip tasks (in
>>>>> above example, "task-c" won't start until "task-a" completes even
>>>>> if we skipped "task-b")
>>>>> 3) We should mark skipped task as `SKIPPED` instead of `SUCCESS`
>>>>> to make it more intuitive
>>>>> 4) The implementation should be easy, clean and low risk
>>>>>
>>>>> Here is my proposed solution (tested locally):
>>>>> Today, Airflow allow user to pass a JSON to the Dagrun as
>>>>> {{dag_run.conf}} when triggering DAG. The idea is, before queuing
>>>>> task instances that satisfies dependences, `scheduler_job.py`
>>>>> (after we make some change) will filter task instances to skip
>>>>> based on `dag_run.conf` user passes in (e.g. {"skip_tasks":
>>>>> ["task-b"]}), then mark them as SKIPPED.
>>>>>
>>>>> Things I would love to discuss:
>>>>> - What do you think about this feature?
>>>>> - What do you think about the proposed solution?
>>>>> - Did I miss anything that you want to discuss?
>>>>> - Is it necessary to introduce a new state (e.g. MANUAL_SKIPPED)
>>>>> to differentiate SKIPPED?
>>>>> Howie
Re: [DISCUSSION] Specify tasks to skip when triggering DAG
Posted by Hongyi Wang <wh...@gmail.com>.
Hi Elad, thank you for your feedback. To answer your question, besides
debugging, another common use case is re-running existing DAGs in an ad-hoc
manner.
For example, as an Airflow user, I sometimes want to trigger an ad-hoc DAG
run. In this run, I want to skip one/more tasks, so the dag run can yield a
different result, or simply complete sooner. As I mentioned in my previous
email, there are other ways to achieve the same goal. But IMHO, neither of
them are easy & flexible enough for an ad-hoc use case.
Does that sound like a reasonable use case? What do you think is the best
approach to solve it? I am happy to discuss more with you.
On Sun, Jan 30, 2022 at 4:45 AM Elad Kalif <el...@apache.org> wrote:
> Can you describe a use case for the requested feature other than
> debugging? This doesn't feel like the right approach to test a specific
> task in a pipeline.
>
> On Fri, Jan 28, 2022 at 11:44 PM Alex Begg <al...@gmail.com> wrote:
>
>> Actually, sorry, you can scratch out some of what I just said, I thought
>> you were talking about clearing states, you are instead referring to
>> triggering a DAG run. That does kind of make sense to have a way to trigger
>> a DAG run but only run specific tasks.
>>
>> On Fri, Jan 28, 2022 at 1:41 PM Alex Begg <al...@gmail.com> wrote:
>>
>>> I believe this is currently possible by just unselecting “downstream”
>>> before you click “Clear” in the UI. It should only clear the one middle
>>> task and not the downstream task(s).
>>>
>>> I would prefer to not have a more detailed UI to allow to skip (or i
>>> want to say “bypass” as “skip” is itself a task state) specific downstream
>>> tasks as it might signal to users that it is ideal to specify tasks to
>>> bypass when in reality it is only something that should be done on occasion
>>> for experiment or troubleshooting as you mention, not a common occurrence.
>>>
>>> What I can agree to though is the list of buttons on the dialog window
>>> to change state of a task is a bit cluttered looking. There probably can be
>>> a better UI/UX for that, but I do think being able to check/uncheck
>>> downstream task is a way to go, that seems like it will be just as
>>> cluttered.
>>>
>>> Alex Begg
>>>
>>> On Fri, Jan 28, 2022 at 11:46 AM Hongyi Wang <wh...@gmail.com> wrote:
>>>
>>>> Hello everyone,
>>>>
>>>> I'd like to propose a new feature in Airflow -- allow users to specify tasks to skip when trigger DAG run.
>>>>
>>>> From our own experience, this feature can be very useful when doing experiments, troubleshooting or re-running existing DAGs. And I believe it can benefit many Airflow users.
>>>>
>>>> To illustrate the use case, I am going to use this example below.
>>>> task-a ☐ -> task-b ☑ -> task-c ☐
>>>>
>>>> Suppose we have a DAG containing 3 tasks. To troubleshoot "task-a" and "task-c", I want to trigger a manual DAG run and skip "task-b" (so I can save time & resource & focus on other two tasks). To do so, today I have two options:
>>>>
>>>> Option 1: Trigger DAG, then manually mark "task-b" as `SUCCESS`
>>>> Option 2: Remove "task-b" from my DAG, then trigger DAG
>>>>
>>>> Neither of the options are great. Option 1 can be troublesome when DAG is large, and there are multiple tasks I want to skip. Option 2 requires change in the DAG file, which is not convenient for just troubleshooting.
>>>>
>>>> Therefore, I would love to discuss how we can provide an easy way for users to skip tasks when triggering DAG.
>>>>
>>>> Things to consider are:
>>>> 1) We should allow user to specify all tasks to skip at once when trigger DAG
>>>> 2) We should retain the dependencies between non-skip tasks (in above example, "task-c" won't start until "task-a" completes even if we skipped "task-b")
>>>> 3) We should mark skipped task as `SKIPPED` instead of `SUCCESS` to make it more intuitive
>>>> 4) The implementation should be easy, clean and low risk
>>>>
>>>> Here is my proposed solution (tested locally):
>>>> Today, Airflow allow user to pass a JSON to the Dagrun as {{dag_run.conf}} when triggering DAG. The idea is, before queuing task instances that satisfies dependences, `scheduler_job.py` (after we make some change) will filter task instances to skip based on `dag_run.conf` user passes in (e.g. {"skip_tasks": ["task-b"]}), then mark them as SKIPPED.
>>>>
>>>> Things I would love to discuss:
>>>> - What do you think about this feature?
>>>> - What do you think about the proposed solution?
>>>> - Did I miss anything that you want to discuss?
>>>> - Is it necessary to introduce a new state (e.g. MANUAL_SKIPPED) to differentiate SKIPPED?
>>>>
>>>> Howie
>>>>
>>>>
Re: [DISCUSSION] Specify tasks to skip when triggering DAG
Posted by Elad Kalif <el...@apache.org>.
Can you describe a use case for the requested feature other than debugging?
This doesn't feel like the right approach to test a specific task in a
pipeline.
On Fri, Jan 28, 2022 at 11:44 PM Alex Begg <al...@gmail.com> wrote:
> Actually, sorry, you can scratch out some of what I just said, I thought
> you were talking about clearing states, you are instead referring to
> triggering a DAG run. That does kind of make sense to have a way to trigger
> a DAG run but only run specific tasks.
>
> On Fri, Jan 28, 2022 at 1:41 PM Alex Begg <al...@gmail.com> wrote:
>
>> I believe this is currently possible by just unselecting “downstream”
>> before you click “Clear” in the UI. It should only clear the one middle
>> task and not the downstream task(s).
>>
>> I would prefer to not have a more detailed UI to allow to skip (or i want
>> to say “bypass” as “skip” is itself a task state) specific downstream tasks
>> as it might signal to users that it is ideal to specify tasks to bypass
>> when in reality it is only something that should be done on occasion for
>> experiment or troubleshooting as you mention, not a common occurrence.
>>
>> What I can agree to though is the list of buttons on the dialog window to
>> change state of a task is a bit cluttered looking. There probably can be a
>> better UI/UX for that, but I do think being able to check/uncheck
>> downstream task is a way to go, that seems like it will be just as
>> cluttered.
>>
>> Alex Begg
>>
>> On Fri, Jan 28, 2022 at 11:46 AM Hongyi Wang <wh...@gmail.com> wrote:
>>
>>> Hello everyone,
>>>
>>> I'd like to propose a new feature in Airflow -- allow users to specify tasks to skip when trigger DAG run.
>>>
>>> From our own experience, this feature can be very useful when doing experiments, troubleshooting or re-running existing DAGs. And I believe it can benefit many Airflow users.
>>>
>>> To illustrate the use case, I am going to use this example below.
>>> task-a ☐ -> task-b ☑ -> task-c ☐
>>>
>>> Suppose we have a DAG containing 3 tasks. To troubleshoot "task-a" and "task-c", I want to trigger a manual DAG run and skip "task-b" (so I can save time & resource & focus on other two tasks). To do so, today I have two options:
>>>
>>> Option 1: Trigger DAG, then manually mark "task-b" as `SUCCESS`
>>> Option 2: Remove "task-b" from my DAG, then trigger DAG
>>>
>>> Neither of the options are great. Option 1 can be troublesome when DAG is large, and there are multiple tasks I want to skip. Option 2 requires change in the DAG file, which is not convenient for just troubleshooting.
>>>
>>> Therefore, I would love to discuss how we can provide an easy way for users to skip tasks when triggering DAG.
>>>
>>> Things to consider are:
>>> 1) We should allow user to specify all tasks to skip at once when trigger DAG
>>> 2) We should retain the dependencies between non-skip tasks (in above example, "task-c" won't start until "task-a" completes even if we skipped "task-b")
>>> 3) We should mark skipped task as `SKIPPED` instead of `SUCCESS` to make it more intuitive
>>> 4) The implementation should be easy, clean and low risk
>>>
>>> Here is my proposed solution (tested locally):
>>> Today, Airflow allow user to pass a JSON to the Dagrun as {{dag_run.conf}} when triggering DAG. The idea is, before queuing task instances that satisfies dependences, `scheduler_job.py` (after we make some change) will filter task instances to skip based on `dag_run.conf` user passes in (e.g. {"skip_tasks": ["task-b"]}), then mark them as SKIPPED.
>>>
>>> Things I would love to discuss:
>>> - What do you think about this feature?
>>> - What do you think about the proposed solution?
>>> - Did I miss anything that you want to discuss?
>>> - Is it necessary to introduce a new state (e.g. MANUAL_SKIPPED) to differentiate SKIPPED?
>>>
>>> Howie
>>>
>>>
Re: [DISCUSSION] Specify tasks to skip when triggering DAG
Posted by Hongyi Wang <wh...@gmail.com>.
No problem Alex. I should make it clearer. Thank you for your clarification.
On Fri, Jan 28, 2022 at 1:44 PM Alex Begg <al...@gmail.com> wrote:
> Actually, sorry, you can scratch out some of what I just said, I thought
> you were talking about clearing states, you are instead referring to
> triggering a DAG run. That does kind of make sense to have a way to trigger
> a DAG run but only run specific tasks.
>
> On Fri, Jan 28, 2022 at 1:41 PM Alex Begg <al...@gmail.com> wrote:
>
>> I believe this is currently possible by just unselecting “downstream”
>> before you click “Clear” in the UI. It should only clear the one middle
>> task and not the downstream task(s).
>>
>> I would prefer to not have a more detailed UI to allow to skip (or i want
>> to say “bypass” as “skip” is itself a task state) specific downstream tasks
>> as it might signal to users that it is ideal to specify tasks to bypass
>> when in reality it is only something that should be done on occasion for
>> experiment or troubleshooting as you mention, not a common occurrence.
>>
>> What I can agree to though is the list of buttons on the dialog window to
>> change state of a task is a bit cluttered looking. There probably can be a
>> better UI/UX for that, but I do think being able to check/uncheck
>> downstream task is a way to go, that seems like it will be just as
>> cluttered.
>>
>> Alex Begg
>>
>> On Fri, Jan 28, 2022 at 11:46 AM Hongyi Wang <wh...@gmail.com> wrote:
>>
>>> Hello everyone,
>>>
>>> I'd like to propose a new feature in Airflow -- allow users to specify tasks to skip when trigger DAG run.
>>>
>>> From our own experience, this feature can be very useful when doing experiments, troubleshooting or re-running existing DAGs. And I believe it can benefit many Airflow users.
>>>
>>> To illustrate the use case, I am going to use this example below.
>>> task-a ☐ -> task-b ☑ -> task-c ☐
>>>
>>> Suppose we have a DAG containing 3 tasks. To troubleshoot "task-a" and "task-c", I want to trigger a manual DAG run and skip "task-b" (so I can save time & resource & focus on other two tasks). To do so, today I have two options:
>>>
>>> Option 1: Trigger DAG, then manually mark "task-b" as `SUCCESS`
>>> Option 2: Remove "task-b" from my DAG, then trigger DAG
>>>
>>> Neither of the options are great. Option 1 can be troublesome when DAG is large, and there are multiple tasks I want to skip. Option 2 requires change in the DAG file, which is not convenient for just troubleshooting.
>>>
>>> Therefore, I would love to discuss how we can provide an easy way for users to skip tasks when triggering DAG.
>>>
>>> Things to consider are:
>>> 1) We should allow user to specify all tasks to skip at once when trigger DAG
>>> 2) We should retain the dependencies between non-skip tasks (in above example, "task-c" won't start until "task-a" completes even if we skipped "task-b")
>>> 3) We should mark skipped task as `SKIPPED` instead of `SUCCESS` to make it more intuitive
>>> 4) The implementation should be easy, clean and low risk
>>>
>>> Here is my proposed solution (tested locally):
>>> Today, Airflow allow user to pass a JSON to the Dagrun as {{dag_run.conf}} when triggering DAG. The idea is, before queuing task instances that satisfies dependences, `scheduler_job.py` (after we make some change) will filter task instances to skip based on `dag_run.conf` user passes in (e.g. {"skip_tasks": ["task-b"]}), then mark them as SKIPPED.
>>>
>>> Things I would love to discuss:
>>> - What do you think about this feature?
>>> - What do you think about the proposed solution?
>>> - Did I miss anything that you want to discuss?
>>> - Is it necessary to introduce a new state (e.g. MANUAL_SKIPPED) to differentiate SKIPPED?
>>>
>>> Howie
>>>
>>>
Re: [DISCUSSION] Specify tasks to skip when triggering DAG
Posted by Alex Begg <al...@gmail.com>.
Actually, sorry, you can scratch out some of what I just said, I thought
you were talking about clearing states, you are instead referring to
triggering a DAG run. That does kind of make sense to have a way to trigger
a DAG run but only run specific tasks.
On Fri, Jan 28, 2022 at 1:41 PM Alex Begg <al...@gmail.com> wrote:
> I believe this is currently possible by just unselecting “downstream”
> before you click “Clear” in the UI. It should only clear the one middle
> task and not the downstream task(s).
>
> I would prefer to not have a more detailed UI to allow to skip (or i want
> to say “bypass” as “skip” is itself a task state) specific downstream tasks
> as it might signal to users that it is ideal to specify tasks to bypass
> when in reality it is only something that should be done on occasion for
> experiment or troubleshooting as you mention, not a common occurrence.
>
> What I can agree to though is the list of buttons on the dialog window to
> change state of a task is a bit cluttered looking. There probably can be a
> better UI/UX for that, but I do think being able to check/uncheck
> downstream task is a way to go, that seems like it will be just as
> cluttered.
>
> Alex Begg
>
> On Fri, Jan 28, 2022 at 11:46 AM Hongyi Wang <wh...@gmail.com> wrote:
>
>> Hello everyone,
>>
>> I'd like to propose a new feature in Airflow -- allow users to specify tasks to skip when trigger DAG run.
>>
>> From our own experience, this feature can be very useful when doing experiments, troubleshooting or re-running existing DAGs. And I believe it can benefit many Airflow users.
>>
>> To illustrate the use case, I am going to use this example below.
>> task-a ☐ -> task-b ☑ -> task-c ☐
>>
>> Suppose we have a DAG containing 3 tasks. To troubleshoot "task-a" and "task-c", I want to trigger a manual DAG run and skip "task-b" (so I can save time & resource & focus on other two tasks). To do so, today I have two options:
>>
>> Option 1: Trigger DAG, then manually mark "task-b" as `SUCCESS`
>> Option 2: Remove "task-b" from my DAG, then trigger DAG
>>
>> Neither of the options are great. Option 1 can be troublesome when DAG is large, and there are multiple tasks I want to skip. Option 2 requires change in the DAG file, which is not convenient for just troubleshooting.
>>
>> Therefore, I would love to discuss how we can provide an easy way for users to skip tasks when triggering DAG.
>>
>> Things to consider are:
>> 1) We should allow user to specify all tasks to skip at once when trigger DAG
>> 2) We should retain the dependencies between non-skip tasks (in above example, "task-c" won't start until "task-a" completes even if we skipped "task-b")
>> 3) We should mark skipped task as `SKIPPED` instead of `SUCCESS` to make it more intuitive
>> 4) The implementation should be easy, clean and low risk
>>
>> Here is my proposed solution (tested locally):
>> Today, Airflow allow user to pass a JSON to the Dagrun as {{dag_run.conf}} when triggering DAG. The idea is, before queuing task instances that satisfies dependences, `scheduler_job.py` (after we make some change) will filter task instances to skip based on `dag_run.conf` user passes in (e.g. {"skip_tasks": ["task-b"]}), then mark them as SKIPPED.
>>
>> Things I would love to discuss:
>> - What do you think about this feature?
>> - What do you think about the proposed solution?
>> - Did I miss anything that you want to discuss?
>> - Is it necessary to introduce a new state (e.g. MANUAL_SKIPPED) to differentiate SKIPPED?
>>
>> Howie
>>
>>
Re: [DISCUSSION] Specify tasks to skip when triggering DAG
Posted by Alex Begg <al...@gmail.com>.
I believe this is currently possible by just unselecting “downstream”
before you click “Clear” in the UI. It should only clear the one middle
task and not the downstream task(s).
I would prefer to not have a more detailed UI to allow to skip (or i want
to say “bypass” as “skip” is itself a task state) specific downstream tasks
as it might signal to users that it is ideal to specify tasks to bypass
when in reality it is only something that should be done on occasion for
experiment or troubleshooting as you mention, not a common occurrence.
What I can agree to though is the list of buttons on the dialog window to
change state of a task is a bit cluttered looking. There probably can be a
better UI/UX for that, but I do think being able to check/uncheck
downstream task is a way to go, that seems like it will be just as
cluttered.
Alex Begg
On Fri, Jan 28, 2022 at 11:46 AM Hongyi Wang <wh...@gmail.com> wrote:
> Hello everyone,
>
> I'd like to propose a new feature in Airflow -- allow users to specify tasks to skip when trigger DAG run.
>
> From our own experience, this feature can be very useful when doing experiments, troubleshooting or re-running existing DAGs. And I believe it can benefit many Airflow users.
>
> To illustrate the use case, I am going to use this example below.
> task-a ☐ -> task-b ☑ -> task-c ☐
>
> Suppose we have a DAG containing 3 tasks. To troubleshoot "task-a" and "task-c", I want to trigger a manual DAG run and skip "task-b" (so I can save time & resource & focus on other two tasks). To do so, today I have two options:
>
> Option 1: Trigger DAG, then manually mark "task-b" as `SUCCESS`
> Option 2: Remove "task-b" from my DAG, then trigger DAG
>
> Neither of the options are great. Option 1 can be troublesome when DAG is large, and there are multiple tasks I want to skip. Option 2 requires change in the DAG file, which is not convenient for just troubleshooting.
>
> Therefore, I would love to discuss how we can provide an easy way for users to skip tasks when triggering DAG.
>
> Things to consider are:
> 1) We should allow user to specify all tasks to skip at once when trigger DAG
> 2) We should retain the dependencies between non-skip tasks (in above example, "task-c" won't start until "task-a" completes even if we skipped "task-b")
> 3) We should mark skipped task as `SKIPPED` instead of `SUCCESS` to make it more intuitive
> 4) The implementation should be easy, clean and low risk
>
> Here is my proposed solution (tested locally):
> Today, Airflow allow user to pass a JSON to the Dagrun as {{dag_run.conf}} when triggering DAG. The idea is, before queuing task instances that satisfies dependences, `scheduler_job.py` (after we make some change) will filter task instances to skip based on `dag_run.conf` user passes in (e.g. {"skip_tasks": ["task-b"]}), then mark them as SKIPPED.
>
> Things I would love to discuss:
> - What do you think about this feature?
> - What do you think about the proposed solution?
> - Did I miss anything that you want to discuss?
> - Is it necessary to introduce a new state (e.g. MANUAL_SKIPPED) to differentiate SKIPPED?
>
> Howie
>
>