You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@airflow.apache.org by Blain David <da...@infrabel.be> on 2024/09/18 16:25:24 UTC
[PROPOSAL] Add streaming support to PartialOperator
Hello,
At our company we have DAG's which have to process lots of paged results (e.g. XComs results from REST endpoints or multiple files from FTP downloads).
The Xcom isn't the issue, as we use a custom provider which allows us to store large data on a Persistent Volume Claim without overloading the database (the Xcom table only holds the reference to the filepath where the XCom data is stored).
The naïve approach of course would be to do a partial on the operator and then expand the paged results, which most of the time is over 1k of tasks.
This of course can become an issue, due to the heavy amount and especially if you have multiple DAG's having the same problem to solve.
Also trying to cancel all those tasks can sometimes be a challenge.
At the time I already brought that issue up in an Airflow Town Hall that it would be nice to be able to "stream" the data instead of having them to spawn all tasks simultaneously before executing them.
So this time I started researching how this could be achieved using the existing Airflow infrastructure, and I implemented a POC which works.
Below a simplified example of a DAG using the stream functionality:
with DAG(
"test_streamed_operator",
default_args=DEFAULT_ARGS,
schedule_interval=timedelta(hours=24),
catchup=False,
) as dag:
distinct_users_ids_task = SQLExecuteQueryOperator(
task_id="distinct_users_ids",
conn_id=SQL_CONN_ID,
sql="SELECT TOP 10 ID FROM USERS",
dag=dag,
)
MSGraphAsyncOperator.partial(
task_id="registered_devices",
conn_id=MSGRAPH_CONN_ID,
url="users/{userId}/registeredDevices",
do_xcom_push=False,
dag=dag,
).stream(path_parameters=distinct_users_ids_task.output.map(lambda user_id: {"userId": user_id [0]}))
The result of the above is that the list of user id's returned by the SQLExecuteQueryOperator will be "streamed" to the path_parameters parameter of the MSGraphAsyncOperator instead of expanding it.
This will result in the "tasks" being executed sequentially instead of launching multiple task instances, this functionality also works with deferable operators like the MSGraphAsyncOperator.
Also it's possible to define a concurrency parameter (is optional), so that the operator can be executed concurrently, without the need to spawn multiple task instance as opposed to expand.
In that case the speed will depend on the number of threads available on the current worker, as no additional workers will be spawned, so this is pure concurrency with the same worker instance.
Of course if your operator is async (deferrable), then this setting won't have any effect.
This could be a nice feature for Airflow 3, as this would solve the scenario when expanding too much task instances is becoming problematic, and "sequential" execution is faster.
And yes I know you can play with the max_active_tasks and max_active_runs parameters, but this facilitates the issue I was trying to solve, especially regarding the sometimes unnecessary creation of dynamic tasks with all the overhead it also leads to.
So I'm curious about your thought about this proposition.
Kind regards,
David
RE: [PROPOSAL] Add streaming support to PartialOperator
Posted by Blain David <da...@infrabel.be>.
The example I give was a simplified version, and also a continuation of another DAG process.
The issue I tried to solve in Airflow here for this case (we have also other use cases where we ran into the same issue) was reading n number of users from MSGraph, which where updated and had to be synchronized in our datawarehouse.
The problem is that for each user, we also then need to update the groups it belongs to, the devices and the licenses for each user, and so on.
Unfortunately, those 3 things I just mentioned need a dedicated MSGraph calls per user, you can't get his information in one call nor even combined with the updated users call, you have to do it all individually.
So in the above example you would get 3 additional calls per updated user, which means 3 extra MSGraph calls. If you have like 1k updated users, that would mean 3k dynamic tasks.
My first approach was using dynamic tasks, but that exploded very quickly as explained above, as each updated user will trigger 3 calls, and users get updated frequently.
For example an updated permission/role for a user will trigger an update, if you have 70k+ users, it can grow quickly.
The original job is running in custom python code using RxPy (https://github.com/ReactiveX/RxPY) which is using the reactive programming methodology,
but we want to step away from it because everything is custom regarding invoking msgraph as well as writing to the database in the code but also the CI/CD involved in maintaining this project.
We want to move away from custom code and have native Airflow jobs, and I personally think this case is perfectly possible in Airflow, at least if we would have the "streaming" option, which I use now and works fine.
-----Original Message-----
From: Daniel Standish <da...@astronomer.io.INVALID>
Sent: Wednesday, September 18, 2024 6:41 PM
To: dev@airflow.apache.org
Subject: Re: [PROPOSAL] Add streaming support to PartialOperator
EXTERNAL MAIL: Indien je de afzender van deze e-mail niet kent en deze niet vertrouwt, klik niet op een link of open geen bijlages. Bij twijfel, stuur deze e-mail als bijlage naar abuse@infrabel.be<ma...@infrabel.be>.
Curious why you want to model this as many tasks, e.g. one page == one task.
Another option would be to handle many pages in one task. And I'm curious what were the factors that led you to split it out more granularly.
Re: [PROPOSAL] Add streaming support to PartialOperator
Posted by Jarek Potiuk <ja...@potiuk.com>.
Oh. I don't think we want to "vote" on it (but I will let David to chime in
because I was mostly guessing what's his expectation and worries are).
On Thu, Oct 17, 2024 at 1:07 AM Vikram Koka <vi...@astronomer.io.invalid>
wrote:
> Hmm, I can think of a different solution to the problem here as well, but I
> could be misunderstanding the problem.
>
> I understand that producing a full AIP may be frustrating, but I don't feel
> confident enough in my understanding that I can vote on this at this time.
> I do think a "light AIP" in the form of a Google doc or equivalent would be
> helpful.
>
>
>
> On Wed, Oct 16, 2024 at 2:58 PM Daniel Standish
> <da...@astronomer.io.invalid> wrote:
>
> > Yeah I agree with Jens here. I think it makes sense to produce an AIP so
> > people can understand the proposal better. We can't really give a thumbs
> > up or down without a proposal. At least it's not like python where you
> > have to implement the whole thing first :)
> >
>
Re: [PROPOSAL] Add streaming support to PartialOperator
Posted by Vikram Koka <vi...@astronomer.io.INVALID>.
Hmm, I can think of a different solution to the problem here as well, but I
could be misunderstanding the problem.
I understand that producing a full AIP may be frustrating, but I don't feel
confident enough in my understanding that I can vote on this at this time.
I do think a "light AIP" in the form of a Google doc or equivalent would be
helpful.
On Wed, Oct 16, 2024 at 2:58 PM Daniel Standish
<da...@astronomer.io.invalid> wrote:
> Yeah I agree with Jens here. I think it makes sense to produce an AIP so
> people can understand the proposal better. We can't really give a thumbs
> up or down without a proposal. At least it's not like python where you
> have to implement the whole thing first :)
>
Re: [PROPOSAL] Add streaming support to PartialOperator
Posted by Daniel Standish <da...@astronomer.io.INVALID>.
Yeah I agree with Jens here. I think it makes sense to produce an AIP so
people can understand the proposal better. We can't really give a thumbs
up or down without a proposal. At least it's not like python where you
have to implement the whole thing first :)
Re: [PROPOSAL] Add streaming support to PartialOperator
Posted by Jarek Potiuk <ja...@potiuk.com>.
I also agree maybe the intentions here are going in the right direction,
but reasoning why something like that is needed is not "streaming" and not
"solve problems with mapped operators".
For me - and maybe I am looking at this proposal through my own looking
glass and completely changing the intention - this might be a beginning of
discussing how we could think about "task affinity" - where we want to run
several dependent tasks (task group?) of Airflow at the same machine -
because we want to share in-memory data between all the tasks. And if I am
derailing the discussion too much - I will shut up, because while I think
we need it eventually (it's been raised multiple times in various
discussions before) but we have pretty much no capacity to implement it now
while doing Airflow 3 (unless someone who does not do Airflow 3 would like
to make a POC of it without big hopes of a lot of meaningful discussion
till February or so).
While the reasoning and use case explained by David initially is different
than that - I think the vision shares interesting properties with what I
think we eventually need: "sub-workflow (task group?)" of airflow tasks
independently visualised in UI, but running together (sequentially or in
parallel as needed) on a single machine where CPU and GPU memory can be
shared between the tasks. That IMHO will open a number of optimisations
that will be needed for some of the low-level machine-learning/AI workflows
that modern building-blocks of the machine learning "ecosystem" allow
(thanks to standards like Apache Arrow - where different tools operate on
the in-memory data with 0-copy overhead basically). And while it's not what
David proposed, really, it also might achieve the same performance gains
that Daniel wants to achieve in the use case of his (almost as a
by-product).
And yes - I strongly agree with Ash that "streaming" is a very bad naming
for either of the two cases - the one that David explained and mine.
But again - I might be derailing the discussion, so I will shut-up if
that's the case.
J.
On Tue, Dec 3, 2024 at 11:43 AM Ash Berlin-Taylor <as...@apache.org> wrote:
> Hi David,
>
> As it stands today I’m -1 to accepting this for a couple of reasons, sorry:
>
> First of all: the implementation looks like it is a “parallel
> implementation” of the scheduler, triggerer. I know that it is in some ways
> a POC only, but there is more for my reasoning.
>
> It feels counter to mapped tasks — the entire point of using mapped tasks
> is to run a single task (or a group of tasks) over a repeated input, and
> have each one be able to run independently, be restarted or retried
> independently and to scale out independently. Right now we loose almost all
> of those benefits.
>
> On the name, “stream” is 100% the wrong word for this concept as it is not
> streaming to process data as it comes in, but the opposite almost, it’s
> batching it all up to run in one group. “Iterate” is much better.
>
> From chatting to David on slack it feels to me like this entire feature is
> built to work around a problem with a large number of mapped tasks. So I’m
> -1 to accepting this in the core project and we should instead spend our
> maintenance effort on improving mapped tasks. It can be maintained as a
> separate operator in a provider out of tree, and if it gains traction we
> could see about bringing it into core as an apache-maintained provider.
>
> In this particular case it also feels like it could be achieved with
> “executor=‘Celery’” + KEDA to scale the worker and get 90% of the same
> behaviour without any changes at all, or as Daniel suggested earlier,
> simply do this in a plain old `for` loop inside a task.
>
> The other idea might be to be able to have a mapped task go directly into
> a triggered — that might also gain the performance you want.
>
> Please let me know if I’ve misunderstood anything about your proposal, and
> sorry to be harsh, but one of the hardest things as a maintainer of an open
> source project is saying no to feature requests.
>
> -ash
>
> > On 6 Nov 2024, at 11:21, Blain David <da...@infrabel.be> wrote:
> >
> > Hello guys,
> >
> > First of all, thank you all for taking your time and giving your
> opinions and insights regarding my proposal. I also think it would indeed
> be better to do an official AIP proposal. I just planted the seed here to
> see how this proposal would be received. I will try to do this as soon as
> possible.
> >
> > Kind regards,
> > David
> >
> > From: Constance Martineau <co...@astronomer.io>
> > Sent: Wednesday, 16 October 2024 23:06
> > To: dev@airflow.apache.org
> > Cc: Blain David <da...@infrabel.be>
> > Subject: Re: [PROPOSAL] Add streaming support to PartialOperator
> >
> > You don't often get email from constance@astronomer.io<mailto:
> constance@astronomer.io>. Learn why this is important<
> https://aka.ms/LearnAboutSenderIdentification>
> >
> >
> > EXTERNAL MAIL: Indien je de afzender van deze e-mail niet kent en deze
> niet vertrouwt, klik niet op een link of open geen bijlages. Bij twijfel,
> stuur deze e-mail als bijlage naar abuse@infrabel.be<mailto:
> abuse@infrabel.be>.
> > That was a lot to read through, and to be honest, it's hard for me to
> tell whether or not Jarek's proposal solves David's problem. However, if
> the debate is whether it's worthwhile or not to provide a first-class way
> for DAG authors to use Operators as part of TaskFlow Tasks, it is.
> >
> > Operators are a major value-add to the Airflow Ecosystem, and we're
> implicitly forcing DAG authors to choose whether they value their DAGs
> being pythonic and simpler to read and reason (Taskflow), or whether they
> value limiting custom code (Traditional Syntax with Operators). There
> should be a first class way to have both, and while it's possible to have
> dependencies between decorated tasks and traditional tasks (and use hooks
> within tasks), you lose a lot of the benefits and it's easier to revert to
> traditional syntax.
> >
> > On Tue, Oct 15, 2024 at 2:46 PM Jens Scheffler
> <j_...@gmx.de.invalid>> wrote:
> > Hi all,
> >
> > thanks for picking-up the discussion. So following the email chain a bit
> > I would recommend to spin an AIP for the implementation. There might be
> > one or multiple cases where this is a cool feature. Still it will add
> > complexity and needs a closer discussion. The best discussion might be
> > on the AIP itself and then once all questions and details are described
> > we still can VOTE on it.
> >
> > @David, can you follow as described in
> >
> https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals
> > ?
> >
> > (I also have another use case in mind and am courious if the propsal
> > would also support this)
> >
> > Jens
> >
> > On 15.10.24 18:24, Daniel Standish wrote:
> >> RE SLAs there was actually a lot of people who chimed in and expressed
> >> concerns with the approach, but no one took the step of actually down
> >> voting it. It's hard to down vote and say no this does not seem right.
> >> And sometimes these things gain a momentum and you don't want to be a
> stick
> >> in the mud, particularly if you don't have a better solution and someone
> >> has spent a lot of time on it. But yeah we should not be so timid about
> >> saying no that we never do it.
> >>
> >> I think I did not really engage with it until substantially later in the
> >> process, wish I could have engaged earlier.
> >>
> >> On the topic of streaming, yeah, I'm trying to do my part to engage in
> this
> >> thread. I don't yet see and understand the value so that's why I
> suggested
> >> fleshing out the proposal in a doc. I'm not ready to give any thumbs up
> yet
> >> cus I'd don't see it. That doesn't mean the value isn't there, just I
> >> don't see it / understand it yet.
> >>
> >> And yeah we're only two people here engaging with this one so, it's
> good if
> >> others could consider the proposal also. But people only have so much
> >> time. And anyway, I think the proposal needs more clarity to be
> >> efficiently and accurately evaluated -- so formalizing it a bit, even if
> >> not precisely an AIP, would help in others to chime in. Really get into
> >> what problem it's solving and why and how.
> >>
> >>
> >>
> >>
> >>
> >>
> >>
> >>
> >>
> >> On Tue, Oct 15, 2024 at 9:05 AM Jarek Potiuk <jarek@potiuk.com<mailto:
> jarek@potiuk.com>> wrote:
> >>
> >>> So I think what David really needs (from you Daniel and others) if is
> the
> >>> idaa sounds right, if it does and we agree it is something that should
> be
> >>> clarified in detail and there are no major blockers to move in this
> >>> direction - this can be turned into detailed proposal with the syntax,
> >>>
> >>> I think we had a long story of some cases (like SLA) where we asked for
> >>> detailed AIPs and then after it has been delivered it turned out that
> the
> >>> idea from the very beginning was not right, but this feedback has been
> >>> missing. SLA feature sufferred from late feedback that "the whole idea
> >>> seems wrong".
> >>>
> >>> I think we should avoid such an approach. If we see that the general
> idea
> >>> is wrong we should give early feedback - and then engage in detailed
> >>> discussion - but without the "I have not paid attention before but the
> >>> whole thing is wrong".
> >>>
> >>> I think David is looking for this kind of confirmation, so that he
> does not
> >>> spend days and weeks on detailing a proposal then was strangled to
> death
> >>> because we did not like the idea in the first place. That's very
> >>> discouraging.
> >>>
> >>> J,
> >>>
> >>> On Tue, Oct 15, 2024 at 6:00 PM Jarek Potiuk <jarek@potiuk.com<mailto:
> jarek@potiuk.com>> wrote:
> >>>
> >>>> It's about the same David's proposal is about stream syntax to run the
> >>>> operators in the task. So those are not two things - this is the
> "idea"
> >>>> (run operators in a loop in a task) and implementation detail (stream
> >>>> syntax).
> >>>>
> >>>> I think at this stage I distilled the idea from the syntax proposal,
> and
> >>>> what we could do in the future is to make sure that syntax is good.
> >>>>
> >>>>
> >>>> J.
> >>>>
> >>>>
> >>>> On Tue, Oct 15, 2024 at 4:11 PM Daniel Standish
> >>>> <daniel.standish@astronomer.io.invalid<mailto:
> daniel.standish@astronomer.io.invalid>> wrote:
> >>>>
> >>>>> I'm still a bit fuzzy on the proposal. It also seems at times like
> you
> >>>>> two
> >>>>> (David and Jarek) are sorta talking about two different things.
> David:
> >>>>> "stream" syntax. Jarek: run operator in a task.
> >>>>>
> >>>>> I would suggest @David maybe just produce a sort of draft AIP maybe
> in
> >>>>> google docs or something and share and interested parties can review
> and
> >>>>> understand better and possibly help shape the direction.
> >>>>>
> >
> > ---------------------------------------------------------------------
> > To unsubscribe, e-mail: dev-unsubscribe@airflow.apache.org<mailto:
> dev-unsubscribe@airflow.apache.org>
> > For additional commands, e-mail: dev-help@airflow.apache.org<mailto:
> dev-help@airflow.apache.org>
>
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: dev-unsubscribe@airflow.apache.org
> For additional commands, e-mail: dev-help@airflow.apache.org
>
>
RE: [PROPOSAL] Add streaming support to PartialOperator
Posted by Blain David <da...@infrabel.be>.
Hello Ash,
First of all, don't apologize for being "harsh" as I don't perceive it that way. I also fully understand that as a maintainer you don't want to add more (and maybe unnecessary) complexity than necessary, especially if it feels like a work around for a specific problem.
As I said, regarding the name, we could indeed rename it as the IteratorOperator for example and rename the stream method the iterate, which would indeed make more sense. I don't have strong opinions on the name because as I said, I got the inspiration from the Java 8 stream API, which there you could also argue if the name they gave to that functionality is the correct one. I kept this name to make the link between the problem that was solved there and the same problem I wanted to solve with operators in Airflow.
I agree with you on all you points, except the last one with the for-loop, because there it's a naïve solution to the problem. How are you going to loop over deferrable operators? First of all, you'll need to understand as a DAG author that the operator will raise a TaskDeferred exception, which means the simple for-loop approach won't work. Then you will realise that in fact you need to invoke the trigger within the TaskDeferred exception used by the deferrable operator, but that code is async, so there you'll have to cope with the event loop and finally, you will probably realise that at the end of that, the deferred operator will also need to call the next method (if defined).
And even worse, you trigger could raise another TaskDeferred exception, which make it even more complex to handle (now you have recursion to handle). For example, the MSGraphAsyncOperator (which I wrote hence why I know it so well) works like that as it implements the producer/consumer pattern through an async triggerer (I think there is also a proposition for that to have a base implementation for this in Airflow as it didn't exist when I implemented it for the MSGraphOperator) when a REST API returns results through multiple pages, that way you don't block your workers waiting for the next response/page.
That's where the StreamedOperator (or IteratorOperator if you prefer) is handy, it encapsulates all this complexity in a generic solution for you which you don't need to worry about and which you can apply on any operator, just like the partial/expand, which will not be possible with custom code in a PythonOperator.
The custom for-loop (which isn't concurrent btw, you could implement it yourself of course as I showed with the snippet in slack), you also don't store the state of each processed item in the for-loop, if the PythonOperator fails, then you'll also need to re-execute it completely. I see it as an alternative way of processing multiple inputs over operators, which is more than a simple for-loop in a PythonOperator which loops over some hook invocations for example, as that is the simple use case.
Also, by looping over the operators within a PythonOperator, you would also violate this concept (https://lists.apache.org/thread/nflt9h6dc5obzztmyqxlpxfs950rtqsq), which I also implemented the PR (https://github.com/apache/airflow/pull/37937) for once Jarek mentioned it as this was also an issue we had to deal with when reviewing DAG's.
And like Joffrey mentioned, as an Airflow infra maintainer, you don't want to see such (very) technical code to solve a technical problem which can't be solved through the expand functionality.
Kind regards,
David
-----Original Message-----
From: Ash Berlin-Taylor <as...@apache.org>
Sent: Tuesday, 3 December 2024 11:44
To: dev@airflow.apache.org
Subject: Re: [PROPOSAL] Add streaming support to PartialOperator
EXTERNAL MAIL: Indien je de afzender van deze e-mail niet kent en deze niet vertrouwt, klik niet op een link of open geen bijlages. Bij twijfel, stuur deze e-mail als bijlage naar abuse@infrabel.be<ma...@infrabel.be>.
Hi David,
As it stands today I’m -1 to accepting this for a couple of reasons, sorry:
First of all: the implementation looks like it is a “parallel implementation” of the scheduler, triggerer. I know that it is in some ways a POC only, but there is more for my reasoning.
It feels counter to mapped tasks — the entire point of using mapped tasks is to run a single task (or a group of tasks) over a repeated input, and have each one be able to run independently, be restarted or retried independently and to scale out independently. Right now we loose almost all of those benefits.
On the name, “stream” is 100% the wrong word for this concept as it is not streaming to process data as it comes in, but the opposite almost, it’s batching it all up to run in one group. “Iterate” is much better.
From chatting to David on slack it feels to me like this entire feature is built to work around a problem with a large number of mapped tasks. So I’m -1 to accepting this in the core project and we should instead spend our maintenance effort on improving mapped tasks. It can be maintained as a separate operator in a provider out of tree, and if it gains traction we could see about bringing it into core as an apache-maintained provider.
In this particular case it also feels like it could be achieved with “executor=‘Celery’” + KEDA to scale the worker and get 90% of the same behaviour without any changes at all, or as Daniel suggested earlier, simply do this in a plain old `for` loop inside a task.
The other idea might be to be able to have a mapped task go directly into a triggered — that might also gain the performance you want.
Please let me know if I’ve misunderstood anything about your proposal, and sorry to be harsh, but one of the hardest things as a maintainer of an open source project is saying no to feature requests.
-ash
> On 6 Nov 2024, at 11:21, Blain David <da...@infrabel.be> wrote:
>
> Hello guys,
>
> First of all, thank you all for taking your time and giving your opinions and insights regarding my proposal. I also think it would indeed be better to do an official AIP proposal. I just planted the seed here to see how this proposal would be received. I will try to do this as soon as possible.
>
> Kind regards,
> David
>
> From: Constance Martineau <co...@astronomer.io>
> Sent: Wednesday, 16 October 2024 23:06
> To: dev@airflow.apache.org
> Cc: Blain David <da...@infrabel.be>
> Subject: Re: [PROPOSAL] Add streaming support to PartialOperator
>
> You don't often get email from
> constance@astronomer.io<ma...@astronomer.io>. Learn why
> this is important<https://aka.ms/LearnAboutSenderIdentification>
>
>
> EXTERNAL MAIL: Indien je de afzender van deze e-mail niet kent en deze niet vertrouwt, klik niet op een link of open geen bijlages. Bij twijfel, stuur deze e-mail als bijlage naar abuse@infrabel.be<ma...@infrabel.be>.
> That was a lot to read through, and to be honest, it's hard for me to tell whether or not Jarek's proposal solves David's problem. However, if the debate is whether it's worthwhile or not to provide a first-class way for DAG authors to use Operators as part of TaskFlow Tasks, it is.
>
> Operators are a major value-add to the Airflow Ecosystem, and we're implicitly forcing DAG authors to choose whether they value their DAGs being pythonic and simpler to read and reason (Taskflow), or whether they value limiting custom code (Traditional Syntax with Operators). There should be a first class way to have both, and while it's possible to have dependencies between decorated tasks and traditional tasks (and use hooks within tasks), you lose a lot of the benefits and it's easier to revert to traditional syntax.
>
> On Tue, Oct 15, 2024 at 2:46 PM Jens Scheffler <j_...@gmx.de.invalid>> wrote:
> Hi all,
>
> thanks for picking-up the discussion. So following the email chain a
> bit I would recommend to spin an AIP for the implementation. There
> might be one or multiple cases where this is a cool feature. Still it
> will add complexity and needs a closer discussion. The best discussion
> might be on the AIP itself and then once all questions and details are
> described we still can VOTE on it.
>
> @David, can you follow as described in
> https://eur01.safelinks.protection.outlook.com/?url=https%3A%2F%2Fcwik
> i.apache.org%2Fconfluence%2Fdisplay%2FAIRFLOW%2FAirflow%2BImprovement%
> 2BProposals&data=05%7C02%7Cdavid.blain%40infrabel.be%7C4c609779ca854eb
> dd47d08dd138764a6%7Cb82bc314ab8e4d6fb18946f02e1f27f2%7C0%7C0%7C6386881
> 94471380080%7CUnknown%7CTWFpbGZsb3d8eyJFbXB0eU1hcGkiOnRydWUsIlYiOiIwLj
> AuMDAwMCIsIlAiOiJXaW4zMiIsIkFOIjoiTWFpbCIsIldUIjoyfQ%3D%3D%7C0%7C%7C%7
> C&sdata=WlsUzIOy5SPV%2F5RFI3zGlGzEwm5U6MlTKRLjxQetzEE%3D&reserved=0
> ?
>
> (I also have another use case in mind and am courious if the propsal
> would also support this)
>
> Jens
>
> On 15.10.24 18:24, Daniel Standish wrote:
>> RE SLAs there was actually a lot of people who chimed in and
>> expressed concerns with the approach, but no one took the step of
>> actually down voting it. It's hard to down vote and say no this does not seem right.
>> And sometimes these things gain a momentum and you don't want to be a
>> stick in the mud, particularly if you don't have a better solution
>> and someone has spent a lot of time on it. But yeah we should not be
>> so timid about saying no that we never do it.
>>
>> I think I did not really engage with it until substantially later in
>> the process, wish I could have engaged earlier.
>>
>> On the topic of streaming, yeah, I'm trying to do my part to engage
>> in this thread. I don't yet see and understand the value so that's
>> why I suggested fleshing out the proposal in a doc. I'm not ready to
>> give any thumbs up yet cus I'd don't see it. That doesn't mean the
>> value isn't there, just I don't see it / understand it yet.
>>
>> And yeah we're only two people here engaging with this one so, it's
>> good if others could consider the proposal also. But people only
>> have so much time. And anyway, I think the proposal needs more
>> clarity to be efficiently and accurately evaluated -- so formalizing
>> it a bit, even if not precisely an AIP, would help in others to chime
>> in. Really get into what problem it's solving and why and how.
>>
>>
>>
>>
>>
>>
>>
>>
>>
>> On Tue, Oct 15, 2024 at 9:05 AM Jarek Potiuk <ja...@potiuk.com>> wrote:
>>
>>> So I think what David really needs (from you Daniel and others) if
>>> is the idaa sounds right, if it does and we agree it is something
>>> that should be clarified in detail and there are no major blockers
>>> to move in this direction - this can be turned into detailed
>>> proposal with the syntax,
>>>
>>> I think we had a long story of some cases (like SLA) where we asked
>>> for detailed AIPs and then after it has been delivered it turned out
>>> that the idea from the very beginning was not right, but this
>>> feedback has been missing. SLA feature sufferred from late feedback
>>> that "the whole idea seems wrong".
>>>
>>> I think we should avoid such an approach. If we see that the general
>>> idea is wrong we should give early feedback - and then engage in
>>> detailed discussion - but without the "I have not paid attention
>>> before but the whole thing is wrong".
>>>
>>> I think David is looking for this kind of confirmation, so that he
>>> does not spend days and weeks on detailing a proposal then was
>>> strangled to death because we did not like the idea in the first
>>> place. That's very discouraging.
>>>
>>> J,
>>>
>>> On Tue, Oct 15, 2024 at 6:00 PM Jarek Potiuk <ja...@potiuk.com>> wrote:
>>>
>>>> It's about the same David's proposal is about stream syntax to run
>>>> the operators in the task. So those are not two things - this is the "idea"
>>>> (run operators in a loop in a task) and implementation detail
>>>> (stream syntax).
>>>>
>>>> I think at this stage I distilled the idea from the syntax
>>>> proposal, and what we could do in the future is to make sure that syntax is good.
>>>>
>>>>
>>>> J.
>>>>
>>>>
>>>> On Tue, Oct 15, 2024 at 4:11 PM Daniel Standish
>>>> <da...@astronomer.io.invalid>> wrote:
>>>>
>>>>> I'm still a bit fuzzy on the proposal. It also seems at times
>>>>> like you two (David and Jarek) are sorta talking about two
>>>>> different things. David:
>>>>> "stream" syntax. Jarek: run operator in a task.
>>>>>
>>>>> I would suggest @David maybe just produce a sort of draft AIP
>>>>> maybe in google docs or something and share and interested parties
>>>>> can review and understand better and possibly help shape the direction.
>>>>>
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail:
> dev-unsubscribe@airflow.apache.org<mailto:dev-unsubscribe@airflow.apac
> he.org> For additional commands, e-mail:
> dev-help@airflow.apache.org<ma...@airflow.apache.org>
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@airflow.apache.org
For additional commands, e-mail: dev-help@airflow.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@airflow.apache.org
For additional commands, e-mail: dev-help@airflow.apache.org
RE: [PROPOSAL] Add streaming support to PartialOperator
Posted by Blain David <da...@infrabel.be>.
I've published a first draft of the AIP proposal on confluence: https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=334760511
The proposal describes the problem I wanted to solved within Airflow 2.x capabilities, how I did it and why. Now we can use this as base on how we can continue further, and what possible alternative and better ways would be possible to achieve the same goal. Any suggestions are welcome.
Kind regards,
David
-----Original Message-----
From: Ash Berlin-Taylor <as...@apache.org>
Sent: Tuesday, 3 December 2024 11:44
To: dev@airflow.apache.org
Subject: Re: [PROPOSAL] Add streaming support to PartialOperator
EXTERNAL MAIL: Indien je de afzender van deze e-mail niet kent en deze niet vertrouwt, klik niet op een link of open geen bijlages. Bij twijfel, stuur deze e-mail als bijlage naar abuse@infrabel.be<ma...@infrabel.be>.
Hi David,
As it stands today I’m -1 to accepting this for a couple of reasons, sorry:
First of all: the implementation looks like it is a “parallel implementation” of the scheduler, triggerer. I know that it is in some ways a POC only, but there is more for my reasoning.
It feels counter to mapped tasks — the entire point of using mapped tasks is to run a single task (or a group of tasks) over a repeated input, and have each one be able to run independently, be restarted or retried independently and to scale out independently. Right now we loose almost all of those benefits.
On the name, “stream” is 100% the wrong word for this concept as it is not streaming to process data as it comes in, but the opposite almost, it’s batching it all up to run in one group. “Iterate” is much better.
From chatting to David on slack it feels to me like this entire feature is built to work around a problem with a large number of mapped tasks. So I’m -1 to accepting this in the core project and we should instead spend our maintenance effort on improving mapped tasks. It can be maintained as a separate operator in a provider out of tree, and if it gains traction we could see about bringing it into core as an apache-maintained provider.
In this particular case it also feels like it could be achieved with “executor=‘Celery’” + KEDA to scale the worker and get 90% of the same behaviour without any changes at all, or as Daniel suggested earlier, simply do this in a plain old `for` loop inside a task.
The other idea might be to be able to have a mapped task go directly into a triggered — that might also gain the performance you want.
Please let me know if I’ve misunderstood anything about your proposal, and sorry to be harsh, but one of the hardest things as a maintainer of an open source project is saying no to feature requests.
-ash
> On 6 Nov 2024, at 11:21, Blain David <da...@infrabel.be> wrote:
>
> Hello guys,
>
> First of all, thank you all for taking your time and giving your opinions and insights regarding my proposal. I also think it would indeed be better to do an official AIP proposal. I just planted the seed here to see how this proposal would be received. I will try to do this as soon as possible.
>
> Kind regards,
> David
>
> From: Constance Martineau <co...@astronomer.io>
> Sent: Wednesday, 16 October 2024 23:06
> To: dev@airflow.apache.org
> Cc: Blain David <da...@infrabel.be>
> Subject: Re: [PROPOSAL] Add streaming support to PartialOperator
>
> You don't often get email from
> constance@astronomer.io<ma...@astronomer.io>. Learn why
> this is important<https://aka.ms/LearnAboutSenderIdentification>
>
>
> EXTERNAL MAIL: Indien je de afzender van deze e-mail niet kent en deze niet vertrouwt, klik niet op een link of open geen bijlages. Bij twijfel, stuur deze e-mail als bijlage naar abuse@infrabel.be<ma...@infrabel.be>.
> That was a lot to read through, and to be honest, it's hard for me to tell whether or not Jarek's proposal solves David's problem. However, if the debate is whether it's worthwhile or not to provide a first-class way for DAG authors to use Operators as part of TaskFlow Tasks, it is.
>
> Operators are a major value-add to the Airflow Ecosystem, and we're implicitly forcing DAG authors to choose whether they value their DAGs being pythonic and simpler to read and reason (Taskflow), or whether they value limiting custom code (Traditional Syntax with Operators). There should be a first class way to have both, and while it's possible to have dependencies between decorated tasks and traditional tasks (and use hooks within tasks), you lose a lot of the benefits and it's easier to revert to traditional syntax.
>
> On Tue, Oct 15, 2024 at 2:46 PM Jens Scheffler <j_...@gmx.de.invalid>> wrote:
> Hi all,
>
> thanks for picking-up the discussion. So following the email chain a
> bit I would recommend to spin an AIP for the implementation. There
> might be one or multiple cases where this is a cool feature. Still it
> will add complexity and needs a closer discussion. The best discussion
> might be on the AIP itself and then once all questions and details are
> described we still can VOTE on it.
>
> @David, can you follow as described in
> https://eur01.safelinks.protection.outlook.com/?url=https%3A%2F%2Fcwik
> i.apache.org%2Fconfluence%2Fdisplay%2FAIRFLOW%2FAirflow%2BImprovement%
> 2BProposals&data=05%7C02%7Cdavid.blain%40infrabel.be%7C4c609779ca854eb
> dd47d08dd138764a6%7Cb82bc314ab8e4d6fb18946f02e1f27f2%7C0%7C0%7C6386881
> 94471380080%7CUnknown%7CTWFpbGZsb3d8eyJFbXB0eU1hcGkiOnRydWUsIlYiOiIwLj
> AuMDAwMCIsIlAiOiJXaW4zMiIsIkFOIjoiTWFpbCIsIldUIjoyfQ%3D%3D%7C0%7C%7C%7
> C&sdata=WlsUzIOy5SPV%2F5RFI3zGlGzEwm5U6MlTKRLjxQetzEE%3D&reserved=0
> ?
>
> (I also have another use case in mind and am courious if the propsal
> would also support this)
>
> Jens
>
> On 15.10.24 18:24, Daniel Standish wrote:
>> RE SLAs there was actually a lot of people who chimed in and
>> expressed concerns with the approach, but no one took the step of
>> actually down voting it. It's hard to down vote and say no this does not seem right.
>> And sometimes these things gain a momentum and you don't want to be a
>> stick in the mud, particularly if you don't have a better solution
>> and someone has spent a lot of time on it. But yeah we should not be
>> so timid about saying no that we never do it.
>>
>> I think I did not really engage with it until substantially later in
>> the process, wish I could have engaged earlier.
>>
>> On the topic of streaming, yeah, I'm trying to do my part to engage
>> in this thread. I don't yet see and understand the value so that's
>> why I suggested fleshing out the proposal in a doc. I'm not ready to
>> give any thumbs up yet cus I'd don't see it. That doesn't mean the
>> value isn't there, just I don't see it / understand it yet.
>>
>> And yeah we're only two people here engaging with this one so, it's
>> good if others could consider the proposal also. But people only
>> have so much time. And anyway, I think the proposal needs more
>> clarity to be efficiently and accurately evaluated -- so formalizing
>> it a bit, even if not precisely an AIP, would help in others to chime
>> in. Really get into what problem it's solving and why and how.
>>
>>
>>
>>
>>
>>
>>
>>
>>
>> On Tue, Oct 15, 2024 at 9:05 AM Jarek Potiuk <ja...@potiuk.com>> wrote:
>>
>>> So I think what David really needs (from you Daniel and others) if
>>> is the idaa sounds right, if it does and we agree it is something
>>> that should be clarified in detail and there are no major blockers
>>> to move in this direction - this can be turned into detailed
>>> proposal with the syntax,
>>>
>>> I think we had a long story of some cases (like SLA) where we asked
>>> for detailed AIPs and then after it has been delivered it turned out
>>> that the idea from the very beginning was not right, but this
>>> feedback has been missing. SLA feature sufferred from late feedback
>>> that "the whole idea seems wrong".
>>>
>>> I think we should avoid such an approach. If we see that the general
>>> idea is wrong we should give early feedback - and then engage in
>>> detailed discussion - but without the "I have not paid attention
>>> before but the whole thing is wrong".
>>>
>>> I think David is looking for this kind of confirmation, so that he
>>> does not spend days and weeks on detailing a proposal then was
>>> strangled to death because we did not like the idea in the first
>>> place. That's very discouraging.
>>>
>>> J,
>>>
>>> On Tue, Oct 15, 2024 at 6:00 PM Jarek Potiuk <ja...@potiuk.com>> wrote:
>>>
>>>> It's about the same David's proposal is about stream syntax to run
>>>> the operators in the task. So those are not two things - this is the "idea"
>>>> (run operators in a loop in a task) and implementation detail
>>>> (stream syntax).
>>>>
>>>> I think at this stage I distilled the idea from the syntax
>>>> proposal, and what we could do in the future is to make sure that syntax is good.
>>>>
>>>>
>>>> J.
>>>>
>>>>
>>>> On Tue, Oct 15, 2024 at 4:11 PM Daniel Standish
>>>> <da...@astronomer.io.invalid>> wrote:
>>>>
>>>>> I'm still a bit fuzzy on the proposal. It also seems at times
>>>>> like you two (David and Jarek) are sorta talking about two
>>>>> different things. David:
>>>>> "stream" syntax. Jarek: run operator in a task.
>>>>>
>>>>> I would suggest @David maybe just produce a sort of draft AIP
>>>>> maybe in google docs or something and share and interested parties
>>>>> can review and understand better and possibly help shape the direction.
>>>>>
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail:
> dev-unsubscribe@airflow.apache.org<mailto:dev-unsubscribe@airflow.apac
> he.org> For additional commands, e-mail:
> dev-help@airflow.apache.org<ma...@airflow.apache.org>
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@airflow.apache.org
For additional commands, e-mail: dev-help@airflow.apache.org
Re: [PROPOSAL] Add streaming support to PartialOperator
Posted by Ash Berlin-Taylor <as...@apache.org>.
Hi David,
As it stands today I’m -1 to accepting this for a couple of reasons, sorry:
First of all: the implementation looks like it is a “parallel implementation” of the scheduler, triggerer. I know that it is in some ways a POC only, but there is more for my reasoning.
It feels counter to mapped tasks — the entire point of using mapped tasks is to run a single task (or a group of tasks) over a repeated input, and have each one be able to run independently, be restarted or retried independently and to scale out independently. Right now we loose almost all of those benefits.
On the name, “stream” is 100% the wrong word for this concept as it is not streaming to process data as it comes in, but the opposite almost, it’s batching it all up to run in one group. “Iterate” is much better.
From chatting to David on slack it feels to me like this entire feature is built to work around a problem with a large number of mapped tasks. So I’m -1 to accepting this in the core project and we should instead spend our maintenance effort on improving mapped tasks. It can be maintained as a separate operator in a provider out of tree, and if it gains traction we could see about bringing it into core as an apache-maintained provider.
In this particular case it also feels like it could be achieved with “executor=‘Celery’” + KEDA to scale the worker and get 90% of the same behaviour without any changes at all, or as Daniel suggested earlier, simply do this in a plain old `for` loop inside a task.
The other idea might be to be able to have a mapped task go directly into a triggered — that might also gain the performance you want.
Please let me know if I’ve misunderstood anything about your proposal, and sorry to be harsh, but one of the hardest things as a maintainer of an open source project is saying no to feature requests.
-ash
> On 6 Nov 2024, at 11:21, Blain David <da...@infrabel.be> wrote:
>
> Hello guys,
>
> First of all, thank you all for taking your time and giving your opinions and insights regarding my proposal. I also think it would indeed be better to do an official AIP proposal. I just planted the seed here to see how this proposal would be received. I will try to do this as soon as possible.
>
> Kind regards,
> David
>
> From: Constance Martineau <co...@astronomer.io>
> Sent: Wednesday, 16 October 2024 23:06
> To: dev@airflow.apache.org
> Cc: Blain David <da...@infrabel.be>
> Subject: Re: [PROPOSAL] Add streaming support to PartialOperator
>
> You don't often get email from constance@astronomer.io<ma...@astronomer.io>. Learn why this is important<https://aka.ms/LearnAboutSenderIdentification>
>
>
> EXTERNAL MAIL: Indien je de afzender van deze e-mail niet kent en deze niet vertrouwt, klik niet op een link of open geen bijlages. Bij twijfel, stuur deze e-mail als bijlage naar abuse@infrabel.be<ma...@infrabel.be>.
> That was a lot to read through, and to be honest, it's hard for me to tell whether or not Jarek's proposal solves David's problem. However, if the debate is whether it's worthwhile or not to provide a first-class way for DAG authors to use Operators as part of TaskFlow Tasks, it is.
>
> Operators are a major value-add to the Airflow Ecosystem, and we're implicitly forcing DAG authors to choose whether they value their DAGs being pythonic and simpler to read and reason (Taskflow), or whether they value limiting custom code (Traditional Syntax with Operators). There should be a first class way to have both, and while it's possible to have dependencies between decorated tasks and traditional tasks (and use hooks within tasks), you lose a lot of the benefits and it's easier to revert to traditional syntax.
>
> On Tue, Oct 15, 2024 at 2:46 PM Jens Scheffler <j_...@gmx.de.invalid>> wrote:
> Hi all,
>
> thanks for picking-up the discussion. So following the email chain a bit
> I would recommend to spin an AIP for the implementation. There might be
> one or multiple cases where this is a cool feature. Still it will add
> complexity and needs a closer discussion. The best discussion might be
> on the AIP itself and then once all questions and details are described
> we still can VOTE on it.
>
> @David, can you follow as described in
> https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals
> ?
>
> (I also have another use case in mind and am courious if the propsal
> would also support this)
>
> Jens
>
> On 15.10.24 18:24, Daniel Standish wrote:
>> RE SLAs there was actually a lot of people who chimed in and expressed
>> concerns with the approach, but no one took the step of actually down
>> voting it. It's hard to down vote and say no this does not seem right.
>> And sometimes these things gain a momentum and you don't want to be a stick
>> in the mud, particularly if you don't have a better solution and someone
>> has spent a lot of time on it. But yeah we should not be so timid about
>> saying no that we never do it.
>>
>> I think I did not really engage with it until substantially later in the
>> process, wish I could have engaged earlier.
>>
>> On the topic of streaming, yeah, I'm trying to do my part to engage in this
>> thread. I don't yet see and understand the value so that's why I suggested
>> fleshing out the proposal in a doc. I'm not ready to give any thumbs up yet
>> cus I'd don't see it. That doesn't mean the value isn't there, just I
>> don't see it / understand it yet.
>>
>> And yeah we're only two people here engaging with this one so, it's good if
>> others could consider the proposal also. But people only have so much
>> time. And anyway, I think the proposal needs more clarity to be
>> efficiently and accurately evaluated -- so formalizing it a bit, even if
>> not precisely an AIP, would help in others to chime in. Really get into
>> what problem it's solving and why and how.
>>
>>
>>
>>
>>
>>
>>
>>
>>
>> On Tue, Oct 15, 2024 at 9:05 AM Jarek Potiuk <ja...@potiuk.com>> wrote:
>>
>>> So I think what David really needs (from you Daniel and others) if is the
>>> idaa sounds right, if it does and we agree it is something that should be
>>> clarified in detail and there are no major blockers to move in this
>>> direction - this can be turned into detailed proposal with the syntax,
>>>
>>> I think we had a long story of some cases (like SLA) where we asked for
>>> detailed AIPs and then after it has been delivered it turned out that the
>>> idea from the very beginning was not right, but this feedback has been
>>> missing. SLA feature sufferred from late feedback that "the whole idea
>>> seems wrong".
>>>
>>> I think we should avoid such an approach. If we see that the general idea
>>> is wrong we should give early feedback - and then engage in detailed
>>> discussion - but without the "I have not paid attention before but the
>>> whole thing is wrong".
>>>
>>> I think David is looking for this kind of confirmation, so that he does not
>>> spend days and weeks on detailing a proposal then was strangled to death
>>> because we did not like the idea in the first place. That's very
>>> discouraging.
>>>
>>> J,
>>>
>>> On Tue, Oct 15, 2024 at 6:00 PM Jarek Potiuk <ja...@potiuk.com>> wrote:
>>>
>>>> It's about the same David's proposal is about stream syntax to run the
>>>> operators in the task. So those are not two things - this is the "idea"
>>>> (run operators in a loop in a task) and implementation detail (stream
>>>> syntax).
>>>>
>>>> I think at this stage I distilled the idea from the syntax proposal, and
>>>> what we could do in the future is to make sure that syntax is good.
>>>>
>>>>
>>>> J.
>>>>
>>>>
>>>> On Tue, Oct 15, 2024 at 4:11 PM Daniel Standish
>>>> <da...@astronomer.io.invalid>> wrote:
>>>>
>>>>> I'm still a bit fuzzy on the proposal. It also seems at times like you
>>>>> two
>>>>> (David and Jarek) are sorta talking about two different things. David:
>>>>> "stream" syntax. Jarek: run operator in a task.
>>>>>
>>>>> I would suggest @David maybe just produce a sort of draft AIP maybe in
>>>>> google docs or something and share and interested parties can review and
>>>>> understand better and possibly help shape the direction.
>>>>>
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: dev-unsubscribe@airflow.apache.org<ma...@airflow.apache.org>
> For additional commands, e-mail: dev-help@airflow.apache.org<ma...@airflow.apache.org>
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@airflow.apache.org
For additional commands, e-mail: dev-help@airflow.apache.org
RE: [PROPOSAL] Add streaming support to PartialOperator
Posted by Blain David <da...@infrabel.be>.
Hello guys,
First of all, thank you all for taking your time and giving your opinions and insights regarding my proposal. I also think it would indeed be better to do an official AIP proposal. I just planted the seed here to see how this proposal would be received. I will try to do this as soon as possible.
Kind regards,
David
From: Constance Martineau <co...@astronomer.io>
Sent: Wednesday, 16 October 2024 23:06
To: dev@airflow.apache.org
Cc: Blain David <da...@infrabel.be>
Subject: Re: [PROPOSAL] Add streaming support to PartialOperator
You don't often get email from constance@astronomer.io<ma...@astronomer.io>. Learn why this is important<https://aka.ms/LearnAboutSenderIdentification>
EXTERNAL MAIL: Indien je de afzender van deze e-mail niet kent en deze niet vertrouwt, klik niet op een link of open geen bijlages. Bij twijfel, stuur deze e-mail als bijlage naar abuse@infrabel.be<ma...@infrabel.be>.
That was a lot to read through, and to be honest, it's hard for me to tell whether or not Jarek's proposal solves David's problem. However, if the debate is whether it's worthwhile or not to provide a first-class way for DAG authors to use Operators as part of TaskFlow Tasks, it is.
Operators are a major value-add to the Airflow Ecosystem, and we're implicitly forcing DAG authors to choose whether they value their DAGs being pythonic and simpler to read and reason (Taskflow), or whether they value limiting custom code (Traditional Syntax with Operators). There should be a first class way to have both, and while it's possible to have dependencies between decorated tasks and traditional tasks (and use hooks within tasks), you lose a lot of the benefits and it's easier to revert to traditional syntax.
On Tue, Oct 15, 2024 at 2:46 PM Jens Scheffler <j_...@gmx.de.invalid>> wrote:
Hi all,
thanks for picking-up the discussion. So following the email chain a bit
I would recommend to spin an AIP for the implementation. There might be
one or multiple cases where this is a cool feature. Still it will add
complexity and needs a closer discussion. The best discussion might be
on the AIP itself and then once all questions and details are described
we still can VOTE on it.
@David, can you follow as described in
https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals
?
(I also have another use case in mind and am courious if the propsal
would also support this)
Jens
On 15.10.24 18:24, Daniel Standish wrote:
> RE SLAs there was actually a lot of people who chimed in and expressed
> concerns with the approach, but no one took the step of actually down
> voting it. It's hard to down vote and say no this does not seem right.
> And sometimes these things gain a momentum and you don't want to be a stick
> in the mud, particularly if you don't have a better solution and someone
> has spent a lot of time on it. But yeah we should not be so timid about
> saying no that we never do it.
>
> I think I did not really engage with it until substantially later in the
> process, wish I could have engaged earlier.
>
> On the topic of streaming, yeah, I'm trying to do my part to engage in this
> thread. I don't yet see and understand the value so that's why I suggested
> fleshing out the proposal in a doc. I'm not ready to give any thumbs up yet
> cus I'd don't see it. That doesn't mean the value isn't there, just I
> don't see it / understand it yet.
>
> And yeah we're only two people here engaging with this one so, it's good if
> others could consider the proposal also. But people only have so much
> time. And anyway, I think the proposal needs more clarity to be
> efficiently and accurately evaluated -- so formalizing it a bit, even if
> not precisely an AIP, would help in others to chime in. Really get into
> what problem it's solving and why and how.
>
>
>
>
>
>
>
>
>
> On Tue, Oct 15, 2024 at 9:05 AM Jarek Potiuk <ja...@potiuk.com>> wrote:
>
>> So I think what David really needs (from you Daniel and others) if is the
>> idaa sounds right, if it does and we agree it is something that should be
>> clarified in detail and there are no major blockers to move in this
>> direction - this can be turned into detailed proposal with the syntax,
>>
>> I think we had a long story of some cases (like SLA) where we asked for
>> detailed AIPs and then after it has been delivered it turned out that the
>> idea from the very beginning was not right, but this feedback has been
>> missing. SLA feature sufferred from late feedback that "the whole idea
>> seems wrong".
>>
>> I think we should avoid such an approach. If we see that the general idea
>> is wrong we should give early feedback - and then engage in detailed
>> discussion - but without the "I have not paid attention before but the
>> whole thing is wrong".
>>
>> I think David is looking for this kind of confirmation, so that he does not
>> spend days and weeks on detailing a proposal then was strangled to death
>> because we did not like the idea in the first place. That's very
>> discouraging.
>>
>> J,
>>
>> On Tue, Oct 15, 2024 at 6:00 PM Jarek Potiuk <ja...@potiuk.com>> wrote:
>>
>>> It's about the same David's proposal is about stream syntax to run the
>>> operators in the task. So those are not two things - this is the "idea"
>>> (run operators in a loop in a task) and implementation detail (stream
>>> syntax).
>>>
>>> I think at this stage I distilled the idea from the syntax proposal, and
>>> what we could do in the future is to make sure that syntax is good.
>>>
>>>
>>> J.
>>>
>>>
>>> On Tue, Oct 15, 2024 at 4:11 PM Daniel Standish
>>> <da...@astronomer.io.invalid>> wrote:
>>>
>>>> I'm still a bit fuzzy on the proposal. It also seems at times like you
>>>> two
>>>> (David and Jarek) are sorta talking about two different things. David:
>>>> "stream" syntax. Jarek: run operator in a task.
>>>>
>>>> I would suggest @David maybe just produce a sort of draft AIP maybe in
>>>> google docs or something and share and interested parties can review and
>>>> understand better and possibly help shape the direction.
>>>>
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@airflow.apache.org<ma...@airflow.apache.org>
For additional commands, e-mail: dev-help@airflow.apache.org<ma...@airflow.apache.org>
Re: [PROPOSAL] Add streaming support to PartialOperator
Posted by Constance Martineau <co...@astronomer.io.INVALID>.
That was a lot to read through, and to be honest, it's hard for me to tell
whether or not Jarek's proposal solves David's problem. However, if the
debate is whether it's worthwhile or not to provide a first-class way for
DAG authors to use Operators as part of TaskFlow Tasks, it is.
Operators are a major value-add to the Airflow Ecosystem, and we're
implicitly forcing DAG authors to choose whether they value their DAGs
being pythonic and simpler to read and reason (Taskflow), or whether they
value limiting custom code (Traditional Syntax with Operators). There
should be a first class way to have both, and while it's possible to have
dependencies between decorated tasks and traditional tasks (and use hooks
within tasks), you lose a lot of the benefits and it's easier to revert to
traditional syntax.
On Tue, Oct 15, 2024 at 2:46 PM Jens Scheffler <j_...@gmx.de.invalid>
wrote:
> Hi all,
>
> thanks for picking-up the discussion. So following the email chain a bit
> I would recommend to spin an AIP for the implementation. There might be
> one or multiple cases where this is a cool feature. Still it will add
> complexity and needs a closer discussion. The best discussion might be
> on the AIP itself and then once all questions and details are described
> we still can VOTE on it.
>
> @David, can you follow as described in
>
> https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals
> ?
>
> (I also have another use case in mind and am courious if the propsal
> would also support this)
>
> Jens
>
> On 15.10.24 18:24, Daniel Standish wrote:
> > RE SLAs there was actually a lot of people who chimed in and expressed
> > concerns with the approach, but no one took the step of actually down
> > voting it. It's hard to down vote and say no this does not seem right.
> > And sometimes these things gain a momentum and you don't want to be a
> stick
> > in the mud, particularly if you don't have a better solution and someone
> > has spent a lot of time on it. But yeah we should not be so timid about
> > saying no that we never do it.
> >
> > I think I did not really engage with it until substantially later in the
> > process, wish I could have engaged earlier.
> >
> > On the topic of streaming, yeah, I'm trying to do my part to engage in
> this
> > thread. I don't yet see and understand the value so that's why I
> suggested
> > fleshing out the proposal in a doc. I'm not ready to give any thumbs up
> yet
> > cus I'd don't see it. That doesn't mean the value isn't there, just I
> > don't see it / understand it yet.
> >
> > And yeah we're only two people here engaging with this one so, it's good
> if
> > others could consider the proposal also. But people only have so much
> > time. And anyway, I think the proposal needs more clarity to be
> > efficiently and accurately evaluated -- so formalizing it a bit, even if
> > not precisely an AIP, would help in others to chime in. Really get into
> > what problem it's solving and why and how.
> >
> >
> >
> >
> >
> >
> >
> >
> >
> > On Tue, Oct 15, 2024 at 9:05 AM Jarek Potiuk <ja...@potiuk.com> wrote:
> >
> >> So I think what David really needs (from you Daniel and others) if is
> the
> >> idaa sounds right, if it does and we agree it is something that should
> be
> >> clarified in detail and there are no major blockers to move in this
> >> direction - this can be turned into detailed proposal with the syntax,
> >>
> >> I think we had a long story of some cases (like SLA) where we asked for
> >> detailed AIPs and then after it has been delivered it turned out that
> the
> >> idea from the very beginning was not right, but this feedback has been
> >> missing. SLA feature sufferred from late feedback that "the whole idea
> >> seems wrong".
> >>
> >> I think we should avoid such an approach. If we see that the general
> idea
> >> is wrong we should give early feedback - and then engage in detailed
> >> discussion - but without the "I have not paid attention before but the
> >> whole thing is wrong".
> >>
> >> I think David is looking for this kind of confirmation, so that he does
> not
> >> spend days and weeks on detailing a proposal then was strangled to death
> >> because we did not like the idea in the first place. That's very
> >> discouraging.
> >>
> >> J,
> >>
> >> On Tue, Oct 15, 2024 at 6:00 PM Jarek Potiuk <ja...@potiuk.com> wrote:
> >>
> >>> It's about the same David's proposal is about stream syntax to run the
> >>> operators in the task. So those are not two things - this is the "idea"
> >>> (run operators in a loop in a task) and implementation detail (stream
> >>> syntax).
> >>>
> >>> I think at this stage I distilled the idea from the syntax proposal,
> and
> >>> what we could do in the future is to make sure that syntax is good.
> >>>
> >>>
> >>> J.
> >>>
> >>>
> >>> On Tue, Oct 15, 2024 at 4:11 PM Daniel Standish
> >>> <da...@astronomer.io.invalid> wrote:
> >>>
> >>>> I'm still a bit fuzzy on the proposal. It also seems at times like
> you
> >>>> two
> >>>> (David and Jarek) are sorta talking about two different things.
> David:
> >>>> "stream" syntax. Jarek: run operator in a task.
> >>>>
> >>>> I would suggest @David maybe just produce a sort of draft AIP maybe in
> >>>> google docs or something and share and interested parties can review
> and
> >>>> understand better and possibly help shape the direction.
> >>>>
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: dev-unsubscribe@airflow.apache.org
> For additional commands, e-mail: dev-help@airflow.apache.org
>
>
Re: [PROPOSAL] Add streaming support to PartialOperator
Posted by Jens Scheffler <j_...@gmx.de.INVALID>.
Hi all,
thanks for picking-up the discussion. So following the email chain a bit
I would recommend to spin an AIP for the implementation. There might be
one or multiple cases where this is a cool feature. Still it will add
complexity and needs a closer discussion. The best discussion might be
on the AIP itself and then once all questions and details are described
we still can VOTE on it.
@David, can you follow as described in
https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals
?
(I also have another use case in mind and am courious if the propsal
would also support this)
Jens
On 15.10.24 18:24, Daniel Standish wrote:
> RE SLAs there was actually a lot of people who chimed in and expressed
> concerns with the approach, but no one took the step of actually down
> voting it. It's hard to down vote and say no this does not seem right.
> And sometimes these things gain a momentum and you don't want to be a stick
> in the mud, particularly if you don't have a better solution and someone
> has spent a lot of time on it. But yeah we should not be so timid about
> saying no that we never do it.
>
> I think I did not really engage with it until substantially later in the
> process, wish I could have engaged earlier.
>
> On the topic of streaming, yeah, I'm trying to do my part to engage in this
> thread. I don't yet see and understand the value so that's why I suggested
> fleshing out the proposal in a doc. I'm not ready to give any thumbs up yet
> cus I'd don't see it. That doesn't mean the value isn't there, just I
> don't see it / understand it yet.
>
> And yeah we're only two people here engaging with this one so, it's good if
> others could consider the proposal also. But people only have so much
> time. And anyway, I think the proposal needs more clarity to be
> efficiently and accurately evaluated -- so formalizing it a bit, even if
> not precisely an AIP, would help in others to chime in. Really get into
> what problem it's solving and why and how.
>
>
>
>
>
>
>
>
>
> On Tue, Oct 15, 2024 at 9:05 AM Jarek Potiuk <ja...@potiuk.com> wrote:
>
>> So I think what David really needs (from you Daniel and others) if is the
>> idaa sounds right, if it does and we agree it is something that should be
>> clarified in detail and there are no major blockers to move in this
>> direction - this can be turned into detailed proposal with the syntax,
>>
>> I think we had a long story of some cases (like SLA) where we asked for
>> detailed AIPs and then after it has been delivered it turned out that the
>> idea from the very beginning was not right, but this feedback has been
>> missing. SLA feature sufferred from late feedback that "the whole idea
>> seems wrong".
>>
>> I think we should avoid such an approach. If we see that the general idea
>> is wrong we should give early feedback - and then engage in detailed
>> discussion - but without the "I have not paid attention before but the
>> whole thing is wrong".
>>
>> I think David is looking for this kind of confirmation, so that he does not
>> spend days and weeks on detailing a proposal then was strangled to death
>> because we did not like the idea in the first place. That's very
>> discouraging.
>>
>> J,
>>
>> On Tue, Oct 15, 2024 at 6:00 PM Jarek Potiuk <ja...@potiuk.com> wrote:
>>
>>> It's about the same David's proposal is about stream syntax to run the
>>> operators in the task. So those are not two things - this is the "idea"
>>> (run operators in a loop in a task) and implementation detail (stream
>>> syntax).
>>>
>>> I think at this stage I distilled the idea from the syntax proposal, and
>>> what we could do in the future is to make sure that syntax is good.
>>>
>>>
>>> J.
>>>
>>>
>>> On Tue, Oct 15, 2024 at 4:11 PM Daniel Standish
>>> <da...@astronomer.io.invalid> wrote:
>>>
>>>> I'm still a bit fuzzy on the proposal. It also seems at times like you
>>>> two
>>>> (David and Jarek) are sorta talking about two different things. David:
>>>> "stream" syntax. Jarek: run operator in a task.
>>>>
>>>> I would suggest @David maybe just produce a sort of draft AIP maybe in
>>>> google docs or something and share and interested parties can review and
>>>> understand better and possibly help shape the direction.
>>>>
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@airflow.apache.org
For additional commands, e-mail: dev-help@airflow.apache.org
Re: [PROPOSAL] Add streaming support to PartialOperator
Posted by Daniel Standish <da...@astronomer.io.INVALID>.
RE SLAs there was actually a lot of people who chimed in and expressed
concerns with the approach, but no one took the step of actually down
voting it. It's hard to down vote and say no this does not seem right.
And sometimes these things gain a momentum and you don't want to be a stick
in the mud, particularly if you don't have a better solution and someone
has spent a lot of time on it. But yeah we should not be so timid about
saying no that we never do it.
I think I did not really engage with it until substantially later in the
process, wish I could have engaged earlier.
On the topic of streaming, yeah, I'm trying to do my part to engage in this
thread. I don't yet see and understand the value so that's why I suggested
fleshing out the proposal in a doc. I'm not ready to give any thumbs up yet
cus I'd don't see it. That doesn't mean the value isn't there, just I
don't see it / understand it yet.
And yeah we're only two people here engaging with this one so, it's good if
others could consider the proposal also. But people only have so much
time. And anyway, I think the proposal needs more clarity to be
efficiently and accurately evaluated -- so formalizing it a bit, even if
not precisely an AIP, would help in others to chime in. Really get into
what problem it's solving and why and how.
On Tue, Oct 15, 2024 at 9:05 AM Jarek Potiuk <ja...@potiuk.com> wrote:
> So I think what David really needs (from you Daniel and others) if is the
> idaa sounds right, if it does and we agree it is something that should be
> clarified in detail and there are no major blockers to move in this
> direction - this can be turned into detailed proposal with the syntax,
>
> I think we had a long story of some cases (like SLA) where we asked for
> detailed AIPs and then after it has been delivered it turned out that the
> idea from the very beginning was not right, but this feedback has been
> missing. SLA feature sufferred from late feedback that "the whole idea
> seems wrong".
>
> I think we should avoid such an approach. If we see that the general idea
> is wrong we should give early feedback - and then engage in detailed
> discussion - but without the "I have not paid attention before but the
> whole thing is wrong".
>
> I think David is looking for this kind of confirmation, so that he does not
> spend days and weeks on detailing a proposal then was strangled to death
> because we did not like the idea in the first place. That's very
> discouraging.
>
> J,
>
> On Tue, Oct 15, 2024 at 6:00 PM Jarek Potiuk <ja...@potiuk.com> wrote:
>
> > It's about the same David's proposal is about stream syntax to run the
> > operators in the task. So those are not two things - this is the "idea"
> > (run operators in a loop in a task) and implementation detail (stream
> > syntax).
> >
> > I think at this stage I distilled the idea from the syntax proposal, and
> > what we could do in the future is to make sure that syntax is good.
> >
> >
> > J.
> >
> >
> > On Tue, Oct 15, 2024 at 4:11 PM Daniel Standish
> > <da...@astronomer.io.invalid> wrote:
> >
> >> I'm still a bit fuzzy on the proposal. It also seems at times like you
> >> two
> >> (David and Jarek) are sorta talking about two different things. David:
> >> "stream" syntax. Jarek: run operator in a task.
> >>
> >> I would suggest @David maybe just produce a sort of draft AIP maybe in
> >> google docs or something and share and interested parties can review and
> >> understand better and possibly help shape the direction.
> >>
> >
>
Re: [PROPOSAL] Add streaming support to PartialOperator
Posted by Jarek Potiuk <ja...@potiuk.com>.
So I think what David really needs (from you Daniel and others) if is the
idaa sounds right, if it does and we agree it is something that should be
clarified in detail and there are no major blockers to move in this
direction - this can be turned into detailed proposal with the syntax,
I think we had a long story of some cases (like SLA) where we asked for
detailed AIPs and then after it has been delivered it turned out that the
idea from the very beginning was not right, but this feedback has been
missing. SLA feature sufferred from late feedback that "the whole idea
seems wrong".
I think we should avoid such an approach. If we see that the general idea
is wrong we should give early feedback - and then engage in detailed
discussion - but without the "I have not paid attention before but the
whole thing is wrong".
I think David is looking for this kind of confirmation, so that he does not
spend days and weeks on detailing a proposal then was strangled to death
because we did not like the idea in the first place. That's very
discouraging.
J,
On Tue, Oct 15, 2024 at 6:00 PM Jarek Potiuk <ja...@potiuk.com> wrote:
> It's about the same David's proposal is about stream syntax to run the
> operators in the task. So those are not two things - this is the "idea"
> (run operators in a loop in a task) and implementation detail (stream
> syntax).
>
> I think at this stage I distilled the idea from the syntax proposal, and
> what we could do in the future is to make sure that syntax is good.
>
>
> J.
>
>
> On Tue, Oct 15, 2024 at 4:11 PM Daniel Standish
> <da...@astronomer.io.invalid> wrote:
>
>> I'm still a bit fuzzy on the proposal. It also seems at times like you
>> two
>> (David and Jarek) are sorta talking about two different things. David:
>> "stream" syntax. Jarek: run operator in a task.
>>
>> I would suggest @David maybe just produce a sort of draft AIP maybe in
>> google docs or something and share and interested parties can review and
>> understand better and possibly help shape the direction.
>>
>
Re: [PROPOSAL] Add streaming support to PartialOperator
Posted by Jarek Potiuk <ja...@potiuk.com>.
It's about the same David's proposal is about stream syntax to run the
operators in the task. So those are not two things - this is the "idea"
(run operators in a loop in a task) and implementation detail (stream
syntax).
I think at this stage I distilled the idea from the syntax proposal, and
what we could do in the future is to make sure that syntax is good.
J.
On Tue, Oct 15, 2024 at 4:11 PM Daniel Standish
<da...@astronomer.io.invalid> wrote:
> I'm still a bit fuzzy on the proposal. It also seems at times like you two
> (David and Jarek) are sorta talking about two different things. David:
> "stream" syntax. Jarek: run operator in a task.
>
> I would suggest @David maybe just produce a sort of draft AIP maybe in
> google docs or something and share and interested parties can review and
> understand better and possibly help shape the direction.
>
Re: [PROPOSAL] Add streaming support to PartialOperator
Posted by Daniel Standish <da...@astronomer.io.INVALID>.
I'm still a bit fuzzy on the proposal. It also seems at times like you two
(David and Jarek) are sorta talking about two different things. David:
"stream" syntax. Jarek: run operator in a task.
I would suggest @David maybe just produce a sort of draft AIP maybe in
google docs or something and share and interested parties can review and
understand better and possibly help shape the direction.
Re: [PROPOSAL] Add streaming support to PartialOperator
Posted by Jarek Potiuk <ja...@potiuk.com>.
I am all for accepting it. Naming etc. can be worked out but I am very much
for adding mutliple optimisations to Airflow. And this one seems not only
easy but also pretty obvious.
Just a comment here.
I spent the last 2 weeks in the Bay Area and Community Over Code in Denver
and I met a lot of people from our industry who are a lot smarter than me
and who's experience and knowledge I tapped into to learn what is happening
in our industry. People from the Apache Software Foundation (various data
engineering, big data, ML/AI related projects), people - my old friends -
Roman Shaposhnik - who is the VP Legal in the foundation - who is currently
running startups in the ML/AI area especially (that is the hot area of
course), Sid Anand - our PMC member, Greg Czajkowski - ex. VP
engineering in Snowflake, ex. Google, ex. Sun/Java engineer and a number of
other people. I also was a "Data Engineering" track lead at Community Over
Code - 2 days, 16 talks, with countless discussions and questions with a
number of Data Engineering tools, libraries, projects, etc.
And there is one theme that I have been hearing over-and-over-and-over
again. We are at the brink of getting into "optimization wars".
Running ML/ LLM/AI workfloads - both training and inference related ones -
is currently terribly, terribly expensive. It's not sustainable in the long
run. And while the ideas about "what solutions should be used for
LLM/ML/AI" are already pretty firm with transformer architecture pretty
much overtaking everything else, everyone is now focusing on running
whatever they are running in far more optimized ways - reusing GPUs for
multiple parallel workflows - so that parts of the models can be loaded
once to GPU and reused, Arrow used as ubiquitous format to store the data
in order to utilize zero-copy Arrow's capabilities and chaining multiple
tools using arrow to use the same data stored not only in memory but as of
recently in GPU, co-locating data-e the centers from multiple clouds closer
to each other in order to save the bandwidth and decrease latency, heck -
even running all the workloads on K8S in the way that you can at any time
move the stable workflows from Clouds to 3x cheaper on-premise K8S cluster.
All that and more - and "optimizing the cost" (which translates to optimize
all resource use) is all the rage now. A lot of people are also working on
optimizing hardware - NVIDIA is currently the top player, but there are
hardware contenders who are working on specialized hardware that will
optimize some of the workfloads 100x (for example dedicated
transformer-only-capable hardware). But it will take many years for the
hardware to get adopted - because it needs quality APIs, compilers and
adoption of the existing solutions and tools, so for now everyone is trying
to optimize their processes on "data engineering" and software level - by
optimizing out all the overhead of the processing introduced by the
pipelines, libraries and tools they run their workflows with.
This will be a theme for the next 2 or 3 years easily - and we have a
chance to shine as the leader in that space - also by making it very easy
and intuitive to optimize workflows that our users currently have.
I think the solution proposed by David, is a very intuitive and easy way
for the users to optimize their workflows. We should not stop there - I
will start another discussion soon about one of the possible things we
should consider implementing (unicorn executor), but I think the more
options we give our users to easily optimise their workflows in Airflow
3.*, the more suitable it will be for some of the new workflows we have in
mind for Airflow 3 (mainly inference related).
J.
On Mon, Oct 7, 2024 at 10:33 AM Blain David <da...@infrabel.be> wrote:
> Exactly Jarek, and yes Daniel, I use that max_active_tis_per_dag parameter
> to control parallelism. The main reason for this solution is not the
> parallelism itself, but to be able to loop within one same task instance
> without overhead of Xcom's and task scheduling, which of course makes it (a
> lot) faster. By default, it will use parallelism in my implementation, but
> sometimes it's a necessity to not do it when doing IO but still want this
> approach as the processing will still be faster than when doing it through
> dynamic task mapping.
>
> I was aware that if that feature would be added, it would have been
> Airflow 3.x anyway. As I mentioned before, I wanted a POC to know if it
> was easily feasible without too much changes and also I didn't wanted to
> wait until we have 3.0. That's why I tried to implement this generic
> approach at our company to see if it:
>
> 1. Would be easily feasible to implement this without too much changes in
> Airflow base code.
> 2. Would offer better performance (always need to measure to know it).
> 3. Would make the DAG code cleaner (less or no custom code required).
>
> Now that I know it's possible and all 3 points were check marked, we could
> think of a better implementation in Airflow 3.x, if this feature would be
> accepted of course.
>
> -----Original Message-----
> From: Jarek Potiuk <ja...@potiuk.com>
> Sent: Saturday, October 5, 2024 12:35 AM
> To: dev@airflow.apache.org
> Subject: Re: [PROPOSAL] Add streaming support to PartialOperator
>
> EXTERNAL MAIL: Indien je de afzender van deze e-mail niet kent en deze
> niet vertrouwt, klik niet op een link of open geen bijlages. Bij twijfel,
> stuur deze e-mail als bijlage naar abuse@infrabel.be<mailto:
> abuse@infrabel.be>.
>
> From the earlier discussions with David - this is also (and mainly) about
> optimisation. Those operators do very little, and when you add total
> overhead that Airflow adds for scheduling and running every task, then it
> turns out that looping such operator's execute in a single interpreter is
> many, many times faster (several orders of magnitude) than running them
> sequentially as tasks.
>
> On Fri, Oct 4, 2024 at 10:16 AM Daniel Standish
> <da...@astronomer.io.invalid> wrote:
>
> > Well, it looks like we do have concurrency control for mapped tasks
> > after all.
> >
> > See max_active_tis_per_dagrun which was added in
> > https://github.com/apache/airflow/pull/29094.
> >
> > So this would allow you to map over your 3000 users in a single run,
> > but process only one at a time (or 5 or 10 at a time). Does that help
> > your use case?
> >
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: dev-unsubscribe@airflow.apache.org
> For additional commands, e-mail: dev-help@airflow.apache.org
>
RE: [PROPOSAL] Add streaming support to PartialOperator
Posted by Blain David <da...@infrabel.be>.
Exactly Jarek, and yes Daniel, I use that max_active_tis_per_dag parameter to control parallelism. The main reason for this solution is not the parallelism itself, but to be able to loop within one same task instance without overhead of Xcom's and task scheduling, which of course makes it (a lot) faster. By default, it will use parallelism in my implementation, but sometimes it's a necessity to not do it when doing IO but still want this approach as the processing will still be faster than when doing it through dynamic task mapping.
I was aware that if that feature would be added, it would have been Airflow 3.x anyway. As I mentioned before, I wanted a POC to know if it was easily feasible without too much changes and also I didn't wanted to wait until we have 3.0. That's why I tried to implement this generic approach at our company to see if it:
1. Would be easily feasible to implement this without too much changes in Airflow base code.
2. Would offer better performance (always need to measure to know it).
3. Would make the DAG code cleaner (less or no custom code required).
Now that I know it's possible and all 3 points were check marked, we could think of a better implementation in Airflow 3.x, if this feature would be accepted of course.
-----Original Message-----
From: Jarek Potiuk <ja...@potiuk.com>
Sent: Saturday, October 5, 2024 12:35 AM
To: dev@airflow.apache.org
Subject: Re: [PROPOSAL] Add streaming support to PartialOperator
EXTERNAL MAIL: Indien je de afzender van deze e-mail niet kent en deze niet vertrouwt, klik niet op een link of open geen bijlages. Bij twijfel, stuur deze e-mail als bijlage naar abuse@infrabel.be<ma...@infrabel.be>.
From the earlier discussions with David - this is also (and mainly) about optimisation. Those operators do very little, and when you add total overhead that Airflow adds for scheduling and running every task, then it turns out that looping such operator's execute in a single interpreter is many, many times faster (several orders of magnitude) than running them sequentially as tasks.
On Fri, Oct 4, 2024 at 10:16 AM Daniel Standish <da...@astronomer.io.invalid> wrote:
> Well, it looks like we do have concurrency control for mapped tasks
> after all.
>
> See max_active_tis_per_dagrun which was added in
> https://github.com/apache/airflow/pull/29094.
>
> So this would allow you to map over your 3000 users in a single run,
> but process only one at a time (or 5 or 10 at a time). Does that help
> your use case?
>
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@airflow.apache.org
For additional commands, e-mail: dev-help@airflow.apache.org
Re: [PROPOSAL] Add streaming support to PartialOperator
Posted by Jarek Potiuk <ja...@potiuk.com>.
From the earlier discussions with David - this is also (and mainly) about
optimisation. Those operators do very little, and when you add total
overhead that Airflow adds for scheduling and running every task, then it
turns out that looping such operator's execute in a single interpreter is
many, many times faster (several orders of magnitude) than running them
sequentially as tasks.
On Fri, Oct 4, 2024 at 10:16 AM Daniel Standish
<da...@astronomer.io.invalid> wrote:
> Well, it looks like we do have concurrency control for mapped tasks after
> all.
>
> See max_active_tis_per_dagrun which was added in
> https://github.com/apache/airflow/pull/29094.
>
> So this would allow you to map over your 3000 users in a single run, but
> process only one at a time (or 5 or 10 at a time). Does that help your use
> case?
>
Re: [PROPOSAL] Add streaming support to PartialOperator
Posted by Daniel Standish <da...@astronomer.io.INVALID>.
Well, it looks like we do have concurrency control for mapped tasks after
all.
See max_active_tis_per_dagrun which was added in
https://github.com/apache/airflow/pull/29094.
So this would allow you to map over your 3000 users in a single run, but
process only one at a time (or 5 or 10 at a time). Does that help your use
case?
Re: [PROPOSAL] Add streaming support to PartialOperator
Posted by Daniel Standish <da...@astronomer.io.INVALID>.
One thing, it would have to be 3.0 since no new features are going into 2.x
anymore AFAIK.
Do I understand correctly that essentially what you want to be able to do
is limit parallelism in mapped task? E.g. is it correct that you
essentially want to do task mapping, but with parallelism=1? Would that be
equivalent? Or is it something more than that? I am sorta surprised if we
don't have a limit like that. Though I suspect DAG.max_active_tasks would
control it at least.
One other thing I'll mention. Prior to 2.9.3, there was a problem that
caused trouble in some mapping scenarios. After a task runs, it will
schedule downstream tasks. (we call this the mini scheduler). Before 2.9.3
the mini scheduler would try to obtain a lock on dag run before running.
If you have 3000 tasks completing at about the same time, naturally this
can cause trouble. So in 2.9.3 we skip mini scheduling if we can't obtain
lock immediately. I wonder if this was part of the trouble you experienced
with mapping.
RE: [PROPOSAL] Add streaming support to PartialOperator
Posted by Blain David <da...@infrabel.be>.
We cleary want to position ourselves as the second group. Not that we don't write python code, as a fact we do, but once we see something could be solved in a generic way and be contributed back to Airflow, we will also propose that. I understand the use of hooks principle, we also do it, when the operator just don't fit in the usage because of too technical difficulties, but as I already told, we want to keep custom code as minimal as possible.
The reasoning behind it is readability and maintenance, and not all the people who will write a DAG are very technical people. The usage of operators is well documented in Airflow/Astronomer, so for those people the step to write a DAG isn't that hard.
My first example in the PR was using the "stream" functionality (maybe the method should be called "iterate" instead) on a PythonOperator, which was actually a bad example, as the looping could be easily done within the task decorated method itself. Then I changed the "simplified" example with the MSGraphAsyncOperator. And there, as also mentioned by Jarek, some operators do validation and post-processing outside of the hook which is also the case with the MSGraphAsyncOperator as many other operators do. This is not because the operator is badly written, it's due to the nature of the DeferredTask and thus a combination between an operator and a triggerer which serves as one async operator, as this is how the deferred task mechanism is designed in Airflow. If you would have to do the same logic in a PythonOperator or a task decorated method, it would become very messy code (as the code is async), you would have to re-invent the whole post-processing e.g. the next_method part done by the operator as a result of the triggerer. That's why I think this functionality could be handy, the complexity of handling async code and managing the parallelism and deferred code mechanism is handled for you, just like the dynamic task mapping does it, but then executed within the same worker concurrently (if wanted, could be sequentially as you can parametrize it). You could also apply the same functionality on a task decorated method, and focus on the business code instead of solving technical issues.
We now at our company have lots of DAG's with custom code to solve those issues, and as I told in the beginning, we want to make our DAG's as clean and simple as possible, avoid code duplication and re-use the Airflow capabilities to the max without re-inventing the wheel. That's why I thought of this solution, and I thought it could be a nice feature in the future for Airflow 3.x. We are already using this in our Airflow instances, the gain in performance is huge and the DAG's are much easier to understand with less technical code, as the problem has been solved in a generic way. That's also why I've implemented it with the functionality available in Airflow 2.x as a POC, but as already mentioned, it would indeed be better to redo it the clean way in Airflow 3.x. That's why now that I know it's possible in Airflow 2.x, I wanted to come back to you guys with this proposition. I also think it's not a good idea to still implement it in Airflow 2.x as Airflow 3.0 is on it's way. I also see this as a new possible feature for Airflow 3.x, which would be an additional argument to do the migration to 3.x. Anyway, as an Airflow user and contributor, I'm willing to help here. For us it doesn't matter, as we can already use it, but it could nice that other Aiflow users could also benefit from this functionality, as I know this topic has been discussed many times.
-----Original Message-----
From: Jarek Potiuk <ja...@potiuk.com>
Sent: Friday, October 4, 2024 5:52 AM
To: dev@airflow.apache.org
Subject: Re: [PROPOSAL] Add streaming support to PartialOperator
EXTERNAL MAIL: Indien je de afzender van deze e-mail niet kent en deze niet vertrouwt, klik niet op een link of open geen bijlages. Bij twijfel, stuur deze e-mail als bijlage naar abuse@infrabel.be<ma...@infrabel.be>.
> why not just do things sequentially in a loop inside of a task?
Yes I think you nailed it - and I think it's just the abstraction you use in this case.
When you loop in the task to do a small thing many times with one of the integrations of Airflow - you could use Hook for that. But - apparently - this abstraction is difficult to discover and possibly sometimes difficult to discover for the user because all they know is "operators" and they do not know what Hooks do.
So .. the natural way to interact with external integration for many of our users is via Operators - so to allow such looping using operators sounds like "follow what is natural for your users".
Basically.- we are not telling the users "Use hooks", but we are following what our users want to do - "use operators" in this case as it feels more natural for them. I think - now when i think of that - it simply shows that we have two kinds of users in this case:
1) those who know and are happy to write custom operators (they will use
hooks)
2) those who are more comfortable in just putting together existing building blocks - i.e. operators (all they know are dag, operators, dependencies - and when they come to composing things they think of task flow as the way to compose the things they know
Clearly -> allowing to use operators in task flow in this mode would respond to the 2nd group of the users.
For me this is kinda model leadership we should do - when you as a leader in a space try to convince others to do things in one way, but pretty much everyone is not following and stubbornly attempt to use the thing you think is wrong, maybe it's a good time to think "well maybe they are right".
J.
On Thu, Oct 3, 2024 at 8:10 PM Daniel Standish <da...@astronomer.io.invalid> wrote:
> The thing i'm having trouble with is that the problem the user, David,
> is trying to solve is basically, that airflow doesn't like super fine-grained
> tasks. Like let's push this to the limit. I run an ecommerce company
> that has 10M visitors per day and each time they visit we update the
> visitor table. I want to run a daily job to process updates. Should
> I model my pipeline as 1 task per customer? Probably not a good idea.
>
> There's a reason e.g. that databases exist and you can do things in a
> set-based way. There seems to be an analogy here with David's example.
> That's why I asked why model it so fine grained. He does not seem to
> want to write a custom operator, but it would seem it's probably a
> good idea here. One way of thinking about the use case is, I want to
> do things sequentially in a loop -- why not just do things
> sequentially in a loop inside of a task?
>
Re: [PROPOSAL] Add streaming support to PartialOperator
Posted by Jarek Potiuk <ja...@potiuk.com>.
> why not just do things sequentially in a loop inside of a task?
Yes I think you nailed it - and I think it's just the abstraction you use
in this case.
When you loop in the task to do a small thing many times with one of the
integrations of Airflow - you could use Hook for that. But - apparently -
this abstraction is difficult to discover and possibly sometimes difficult
to discover for the user because all they know is "operators" and they do
not know what Hooks do.
So .. the natural way to interact with external integration for many of our
users is via Operators - so to allow such looping using operators sounds
like "follow what is natural for your users".
Basically.- we are not telling the users "Use hooks", but we are following
what our users want to do - "use operators" in this case as it feels more
natural for them. I think - now when i think of that - it simply shows that
we have two kinds of users in this case:
1) those who know and are happy to write custom operators (they will use
hooks)
2) those who are more comfortable in just putting together
existing building blocks - i.e. operators (all they know are dag,
operators, dependencies - and when they come to composing things they think
of task flow as the way to compose the things they know
Clearly -> allowing to use operators in task flow in this mode would
respond to the 2nd group of the users.
For me this is kinda model leadership we should do - when you as a leader
in a space try to convince others to do things in one way, but pretty much
everyone is not following and stubbornly attempt to use the thing you think
is wrong, maybe it's a good time to think "well maybe they are right".
J.
On Thu, Oct 3, 2024 at 8:10 PM Daniel Standish
<da...@astronomer.io.invalid> wrote:
> The thing i'm having trouble with is that the problem the user, David, is
> trying to solve is basically, that airflow doesn't like super fine-grained
> tasks. Like let's push this to the limit. I run an ecommerce company
> that has 10M visitors per day and each time they visit we update the
> visitor table. I want to run a daily job to process updates. Should I
> model my pipeline as 1 task per customer? Probably not a good idea.
>
> There's a reason e.g. that databases exist and you can do things in a
> set-based way. There seems to be an analogy here with David's example.
> That's why I asked why model it so fine grained. He does not seem to want
> to write a custom operator, but it would seem it's probably a good idea
> here. One way of thinking about the use case is, I want to do things
> sequentially in a loop -- why not just do things sequentially in a loop
> inside of a task?
>
Re: [PROPOSAL] Add streaming support to PartialOperator
Posted by Daniel Standish <da...@astronomer.io.INVALID>.
The thing i'm having trouble with is that the problem the user, David, is
trying to solve is basically, that airflow doesn't like super fine-grained
tasks. Like let's push this to the limit. I run an ecommerce company
that has 10M visitors per day and each time they visit we update the
visitor table. I want to run a daily job to process updates. Should I
model my pipeline as 1 task per customer? Probably not a good idea.
There's a reason e.g. that databases exist and you can do things in a
set-based way. There seems to be an analogy here with David's example.
That's why I asked why model it so fine grained. He does not seem to want
to write a custom operator, but it would seem it's probably a good idea
here. One way of thinking about the use case is, I want to do things
sequentially in a loop -- why not just do things sequentially in a loop
inside of a task?
Re: [PROPOSAL] Add streaming support to PartialOperator
Posted by Jarek Potiuk <ja...@potiuk.com>.
Often hooks do a lot of validation -> Often operators do a lot of
validation
On Thu, Oct 3, 2024 at 7:50 PM Jarek Potiuk <ja...@potiuk.com> wrote:
> I think this is very similar to past discussions that we had about
> allowing operators to be used in task flow as a "first class citizen".
> https://lists.apache.org/thread/nflt9h6dc5obzztmyqxlpxfs950rtqsq
> I re-read the original discussion and ...
>
> In theory It sounds like you should be able to do the same using Hooks -
> you should not need to refer to operators. That's (in theory) what the
> hooks are for.
>
> But one of my favourite sayings ....
>
> In theory practice is the same as theory, but in practice it's not.
>
> It definitely looks like our users are not really "aware" of hook
> capability and in fact a number of implemented operators, even if they are
> theoretically thin wrappers around Hooks functionality, the main benefit of
> it (i.e. Hook reusability) is not really materializing. Often hooks do a
> lot of validation and some pre-processing that would otherwise have to be
> copied to Task Flow operators to make the hooks really useful.
>
> For me that sounds like a "design smell" that we could also get rid of in
> Airflow 3 if we make operators runnable directly from task flow - with all
> the bells and whistles including automated jinja template pre-processing
> etc (as discussed in the original thread). I think we have great
> opportunity with Airflow 3 to introduce this and pretty much get rid of the
> notion or even hint that Hooks are the reusability components for DAG
> authors (they would still provide reusability for Operator authors though
> as they do today) - and instead we could promote "Airflow 3 can now mix and
> match operators in Task Flow. So a similar approach as you proposed David
> would be one of the ways to do it (with the exception that I really do not
> like a "streaming" name - which as David mentioned in a Slack conversation
> - direct Java rip-off.
>
> Also following the slack conversation - I know David is eager to do it in
> the way that it will be Airflow 2 compatible, but I think making it Airflow
> 3 feature would be much, much, much more powerful. Even if we **could** do
> it for Airflow 2, it does not necessarily mean we **should** do it.
> Implementing it for Airflow 2 means that any problems and fixes would have
> to be backported and implemented in both etc. etc. - also being able to
> announce it as a "feature" of Airflow 3 "Hey, you finally can mix and match
> operators in Airflow Dag tasks" is a very cool incentive for people to
> migrate.
>
> I would love to hear what others think.
>
> J.
>
>
>
Re: [PROPOSAL] Add streaming support to PartialOperator
Posted by Jarek Potiuk <ja...@potiuk.com>.
I think this is very similar to past discussions that we had about allowing
operators to be used in task flow as a "first class citizen".
https://lists.apache.org/thread/nflt9h6dc5obzztmyqxlpxfs950rtqsq
I re-read the original discussion and ...
In theory It sounds like you should be able to do the same using Hooks -
you should not need to refer to operators. That's (in theory) what the
hooks are for.
But one of my favourite sayings ....
In theory practice is the same as theory, but in practice it's not.
It definitely looks like our users are not really "aware" of hook
capability and in fact a number of implemented operators, even if they are
theoretically thin wrappers around Hooks functionality, the main benefit of
it (i.e. Hook reusability) is not really materializing. Often hooks do a
lot of validation and some pre-processing that would otherwise have to be
copied to Task Flow operators to make the hooks really useful.
For me that sounds like a "design smell" that we could also get rid of in
Airflow 3 if we make operators runnable directly from task flow - with all
the bells and whistles including automated jinja template pre-processing
etc (as discussed in the original thread). I think we have great
opportunity with Airflow 3 to introduce this and pretty much get rid of the
notion or even hint that Hooks are the reusability components for DAG
authors (they would still provide reusability for Operator authors though
as they do today) - and instead we could promote "Airflow 3 can now mix and
match operators in Task Flow. So a similar approach as you proposed David
would be one of the ways to do it (with the exception that I really do not
like a "streaming" name - which as David mentioned in a Slack conversation
- direct Java rip-off.
Also following the slack conversation - I know David is eager to do it in
the way that it will be Airflow 2 compatible, but I think making it Airflow
3 feature would be much, much, much more powerful. Even if we **could** do
it for Airflow 2, it does not necessarily mean we **should** do it.
Implementing it for Airflow 2 means that any problems and fixes would have
to be backported and implemented in both etc. etc. - also being able to
announce it as a "feature" of Airflow 3 "Hey, you finally can mix and match
operators in Airflow Dag tasks" is a very cool incentive for people to
migrate.
I would love to hear what others think.
J.
RE: [PROPOSAL] Add streaming support to PartialOperator
Posted by Blain David <da...@infrabel.be>.
The MSGraphAsyncOperator already handles multi paged responses internally using the defer mechanism, but the result of this are indexed XComs, which means you have to expand over them to process it. We sometimes also use jq in combination with BashOperator to merge those multiple JSON responses so we can handle it in one task, but in some situations, you just can't do that as explained in my previous example.
-----Original Message-----
From: Daniel Standish <da...@astronomer.io.INVALID>
Sent: Wednesday, September 18, 2024 6:41 PM
To: dev@airflow.apache.org
Subject: Re: [PROPOSAL] Add streaming support to PartialOperator
EXTERNAL MAIL: Indien je de afzender van deze e-mail niet kent en deze niet vertrouwt, klik niet op een link of open geen bijlages. Bij twijfel, stuur deze e-mail als bijlage naar abuse@infrabel.be<ma...@infrabel.be>.
Curious why you want to model this as many tasks, e.g. one page == one task.
Another option would be to handle many pages in one task. And I'm curious what were the factors that led you to split it out more granularly.
Re: [PROPOSAL] Add streaming support to PartialOperator
Posted by Daniel Standish <da...@astronomer.io.INVALID>.
Curious why you want to model this as many tasks, e.g. one page == one task.
Another option would be to handle many pages in one task. And I'm curious
what were the factors that led you to split it out more granularly.