Re: [DISCUSS] AIP-12 Persist DAG into DB

[Kevin Yang <yr...@gmail.com>]

Hi all,
When I was preparing some work related to this AIP I found something very
concerning. I noticed this JIRA ticket
<https://issues.apache.org/jira/browse/AIRFLOW-3562> is trying to remove
the dependency of dagbag from webserver, which is awesome--we wanted badly
but never got to start work on. However when I looked at some subtasks of
it, which try to remove dagbag dependency from each endpoint, I found the
way we remove the dependency of dagbag is not very ideal. For example this
PR <https://github.com/apache/airflow/pull/4867/files> will require us to
parse the dag file each time we hit the endpoint.

If we go down this path, we indeed can get rid of the dagbag dependency
easily, but we will have to 1. increase the DB load( not too concerning at
the moment ), 2. wait the DAG file to be parsed before getting the page
back, potentially multiple times. DAG file can sometimes take quite a while
to parse, e.g. we have some framework DAG files generating large number of
DAGs from some static config files or even jupyter notebooks and they can
take 30+ seconds to parse. Yes we don't like large DAG files but people do
see the beauty of code as config and sometimes heavily abuseleverage it.
Assuming all users have the same nice small python file that can be parsed
fast, I'm still a bit worried about this approach. Continuing on this path
means we've chosen DagModel to be the serialized representation of DAG and
DB columns to hold different properties--it can be one candidate but I
don't know if we should settle on that now. I would personally prefer a
more compact, e.g. JSON5, and easy to scale representation( such that
serializing new fields != DB upgrade).

In my imagination we would have to collect the list of dynamic features
depending on unserializable fields of a DAG and start a discussion/vote on
dropping support of them( I'm working on this but if anyone has already
done so please take over), decide on the serialized representation of a DAG
and then replace dagbag with it in webserver. Per previous discussion and
some offline discussions with Dan, one future of DAG serialization that I
like would look similar to this:
[image: airflow_new_arch.jpg]
We can still discuss/vote which approach we want to take but I don't want
the door to above design to be shut right now or we have to spend a lot
effort switch path later.

Bas and Peter, I'm very sorry to extend the discussion but I do think this
is tightly related to the AIP and PRs behind it. And my sincere apology for
bringing this up so late( I only pull the open PR list occasionally, if
there's a way to subscribe to new PR event I'd love to know how).

Cheers,
Kevin Y

On Thu, Feb 28, 2019 at 1:36 PM Peter van t Hof <pj...@gmail.com>
wrote:

> Hi all,
>
> Just some comments one the point Bolke dit give in relation of my PR.
>
> At first, the main focus is: making the webserver stateless.
>
> > 1) Make the webserver stateless: needs the graph of the *current* dag
>
> This is the main goal but for this a lot more PR’s will be coming once my
> current is merged. For edges and graph view this is covered in my PR
> already.
>
> > 2) Version dags: for consistency mainly and not requiring parsing of the
> > dag on every loop
>
> In my PR the historical graphs will be stored for each DagRun. This means
> that you can see if an older DagRun was the same graph structure, even if
> some tasks does not exists anymore in the current graph. Especially for
> dynamic DAG’s this is very useful.
>
> > 3) Make the scheduler not require DAG files. This could be done if the
> > edges contain all information when to trigger the next task. We can then
> > have event driven dag parsing outside of the scheduler loop, ie. by the
> > cli. Storage can also be somewhere else (git, artifactory, filesystem,
> > whatever).
>
> The scheduler is almost untouched in this PR. The only thing that is added
> is that this edges are saved to the database but the scheduling itself
> din’t change. The scheduler depends now still on the DAG object.
>
> > 4) Fully serialise the dag so it becomes transferable to workers
>
> It nice to see that people has a lot of idea’s about this. But as Fokko
> already mentioned this is out of scope for the issue what we are trying to
> solve. I also have some idea’s about this but I like to limit this PR/AIP
> to the webserver.
>
> For now my PR does solve 1 and 2 and the rest of the behaviour (like
> scheduling) is untouched.
>
> Gr,
> Peter
>
>

Re: Pods not being TERMinated properly for K8s airflow

[Jarek Potiuk <Ja...@polidea.com>]

One small comment - in my current WIP multi-layered Docker file AIP-7 (
https://github.com/apache/airflow/pull/4543) I used dumb-init (developed by
Yelp) rather than tini. It's a bit better for pythonic world as it can be
installed from PIP:
'dumb-init>=1.2.2',
 (
https://github.com/apache/airflow/pull/4543/files#diff-2eeaed663bd0d25b7e608891384b7298R311
),

and the usage is  also rather straightforward:

ENTRYPOINT ["/usr/local/bin/dumb-init", "--", "/entrypoint.sh"]

(
https://github.com/apache/airflow/pull/4543/files#diff-3254677a7917c6c01f55212f86c57fbfR361
)

One small caveat I noticed so far with using proper "init" is that when you
want to run interactive shell, you need to override --entrypoint -
otherwise the shell has no terminal to use and complains.

J.


On Tue, Mar 12, 2019 at 4:41 AM Daniel Imberman <da...@gmail.com>
wrote:

> Oh yikes. Thanks for catching this @Jarek! @Greg Neiheisel
> <gr...@astronomer.io>  were there a lot of caveats to deploying from tini
> or
> would this be a fairly straightforward fix?
>
> On Sun, Mar 10, 2019 at 5:13 PM Greg Neiheisel <gr...@astronomer.io> wrote:
>
> > I can confirm that wrapping airflow with tini/dumb-init works pretty
> well.
> > We've been relying on it at Astronomer for the past year. We run
> > exclusively on k8s and are restarting airflow pods very frequently.
> Here's
> > our production 1.10.2 image using tini on alpine, for example
> >
> >
> https://github.com/astronomer/astronomer/blob/master/docker/airflow/1.10.2/Dockerfile#L102
> > .
> >
> > On Sun, Mar 10, 2019 at 3:49 PM Jarek Potiuk <Ja...@polidea.com>
> > wrote:
> >
> > > NOTE! I changed the subject to not pollute the AIP-12 thread
> > >
> > > Fokko,
> > >
> > > I think I know why TERM signals do not work in the current POD
> > > implementation. I already experienced that several times  - dockerized
> > app
> > > not receiving TERM signal. The reason was always the same. It is not a
> > bug
> > > actually - it is expected behaviour in case your ENTRYPOINT is in the
> > SHELL
> > > form ("/binary arg1 arg2") or when you use shell script as ENTRYPOINT
> > first
> > > argument in [ "shell script", "arg" ] form.
> > >
> > > In those cases "shell" process becomes the "0" init process.
> > Unfortunately
> > > shell process is not prepared to do all the stuff that proper init
> > process
> > > should be doing:
> > >
> > > * Inherit orphaned child processes
> > > * reap them
> > > * Handle and propagate signals properly
> > > * Wait until all subprocesses are terminated before terminating itself
> > >
> > > What happens is that the TERM signal just kills the init "shell"
> process
> > > but, then the signal does not reach any of its children and the
> container
> > > continues to run. It's well known problem in docker world and there
> are a
> > > number of solutions (including exec-ing from shell or - better - using
> a
> > > dumb-init/tini in your ENTRYPOINT- very tiny "proper" init
> > implementations
> > > that do what they should do.
> > >
> > > You can read more for example here:
> > > https://hynek.me/articles/docker-signals
> > > .
> > >
> > > J.
> > >
> > > On Sun, Mar 10, 2019 at 7:29 PM Driesprong, Fokko <fokko@driesprong.frl
> >
> > > wrote:
> > >
> > > Ps. Jarek, interesting idea. It shouldn't be too hard to make Airflow
> > more
> > > > k8s native. You could package your dags within your container, and
> do a
> > > > rolling update. Add the DAGs as the last layer, and then point the
> DAGs
> > > > folder to the same location. The hard part here is that you need to
> > > > gracefuly restart the workers. Currently AFAIK the signals given to
> the
> > > pod
> > > > aren't respected. So when the scheduler/webserver/worker receives a
> > > > SIGTERM, it should stop the jobs nicely and then exit the container,
> > > before
> > > > k8s kills the container using a SIGKILL.  This will be challenging
> with
> > > the
> > > > workers, which they are potentially long-running. Maybe stop kicking
> > off
> > > > new jobs, and let the old ones finish, will be good enough, but then
> we
> > > > need to increase the standard kill timeout substantially. Having this
> > > would
> > > > also enable the autoscaling of the workers
> > > >
> > >
> > >
> > > --
> > >
> > > Jarek Potiuk
> > > Polidea <https://www.polidea.com/> | Principal Software Engineer
> > >
> > > M: +48 660 796 129 <+48660796129>
> > > E: jarek.potiuk@polidea.com
> > >
> >
> >
> > --
> > *Greg Neiheisel* / CTO Astronomer.io
> >
>


-- 

Jarek Potiuk
Polidea <https://www.polidea.com/> | Principal Software Engineer

M: +48 660 796 129 <+48660796129>
E: jarek.potiuk@polidea.com

Re: Pods not being TERMinated properly for K8s airflow

[Daniel Imberman <da...@gmail.com>]

Oh yikes. Thanks for catching this @Jarek! @Greg Neiheisel
<gr...@astronomer.io>  were there a lot of caveats to deploying from tini or
would this be a fairly straightforward fix?

On Sun, Mar 10, 2019 at 5:13 PM Greg Neiheisel <gr...@astronomer.io> wrote:

> I can confirm that wrapping airflow with tini/dumb-init works pretty well.
> We've been relying on it at Astronomer for the past year. We run
> exclusively on k8s and are restarting airflow pods very frequently. Here's
> our production 1.10.2 image using tini on alpine, for example
>
> https://github.com/astronomer/astronomer/blob/master/docker/airflow/1.10.2/Dockerfile#L102
> .
>
> On Sun, Mar 10, 2019 at 3:49 PM Jarek Potiuk <Ja...@polidea.com>
> wrote:
>
> > NOTE! I changed the subject to not pollute the AIP-12 thread
> >
> > Fokko,
> >
> > I think I know why TERM signals do not work in the current POD
> > implementation. I already experienced that several times  - dockerized
> app
> > not receiving TERM signal. The reason was always the same. It is not a
> bug
> > actually - it is expected behaviour in case your ENTRYPOINT is in the
> SHELL
> > form ("/binary arg1 arg2") or when you use shell script as ENTRYPOINT
> first
> > argument in [ "shell script", "arg" ] form.
> >
> > In those cases "shell" process becomes the "0" init process.
> Unfortunately
> > shell process is not prepared to do all the stuff that proper init
> process
> > should be doing:
> >
> > * Inherit orphaned child processes
> > * reap them
> > * Handle and propagate signals properly
> > * Wait until all subprocesses are terminated before terminating itself
> >
> > What happens is that the TERM signal just kills the init "shell" process
> > but, then the signal does not reach any of its children and the container
> > continues to run. It's well known problem in docker world and there are a
> > number of solutions (including exec-ing from shell or - better - using a
> > dumb-init/tini in your ENTRYPOINT- very tiny "proper" init
> implementations
> > that do what they should do.
> >
> > You can read more for example here:
> > https://hynek.me/articles/docker-signals
> > .
> >
> > J.
> >
> > On Sun, Mar 10, 2019 at 7:29 PM Driesprong, Fokko <fo...@driesprong.frl>
> > wrote:
> >
> > Ps. Jarek, interesting idea. It shouldn't be too hard to make Airflow
> more
> > > k8s native. You could package your dags within your container, and do a
> > > rolling update. Add the DAGs as the last layer, and then point the DAGs
> > > folder to the same location. The hard part here is that you need to
> > > gracefuly restart the workers. Currently AFAIK the signals given to the
> > pod
> > > aren't respected. So when the scheduler/webserver/worker receives a
> > > SIGTERM, it should stop the jobs nicely and then exit the container,
> > before
> > > k8s kills the container using a SIGKILL.  This will be challenging with
> > the
> > > workers, which they are potentially long-running. Maybe stop kicking
> off
> > > new jobs, and let the old ones finish, will be good enough, but then we
> > > need to increase the standard kill timeout substantially. Having this
> > would
> > > also enable the autoscaling of the workers
> > >
> >
> >
> > --
> >
> > Jarek Potiuk
> > Polidea <https://www.polidea.com/> | Principal Software Engineer
> >
> > M: +48 660 796 129 <+48660796129>
> > E: jarek.potiuk@polidea.com
> >
>
>
> --
> *Greg Neiheisel* / CTO Astronomer.io
>

Re: Pods not being TERMinated properly for K8s airflow

[Greg Neiheisel <gr...@astronomer.io>]

I can confirm that wrapping airflow with tini/dumb-init works pretty well.
We've been relying on it at Astronomer for the past year. We run
exclusively on k8s and are restarting airflow pods very frequently. Here's
our production 1.10.2 image using tini on alpine, for example
https://github.com/astronomer/astronomer/blob/master/docker/airflow/1.10.2/Dockerfile#L102
.

On Sun, Mar 10, 2019 at 3:49 PM Jarek Potiuk <Ja...@polidea.com>
wrote:

> NOTE! I changed the subject to not pollute the AIP-12 thread
>
> Fokko,
>
> I think I know why TERM signals do not work in the current POD
> implementation. I already experienced that several times  - dockerized app
> not receiving TERM signal. The reason was always the same. It is not a bug
> actually - it is expected behaviour in case your ENTRYPOINT is in the SHELL
> form ("/binary arg1 arg2") or when you use shell script as ENTRYPOINT first
> argument in [ "shell script", "arg" ] form.
>
> In those cases "shell" process becomes the "0" init process. Unfortunately
> shell process is not prepared to do all the stuff that proper init process
> should be doing:
>
> * Inherit orphaned child processes
> * reap them
> * Handle and propagate signals properly
> * Wait until all subprocesses are terminated before terminating itself
>
> What happens is that the TERM signal just kills the init "shell" process
> but, then the signal does not reach any of its children and the container
> continues to run. It's well known problem in docker world and there are a
> number of solutions (including exec-ing from shell or - better - using a
> dumb-init/tini in your ENTRYPOINT- very tiny "proper" init implementations
> that do what they should do.
>
> You can read more for example here:
> https://hynek.me/articles/docker-signals
> .
>
> J.
>
> On Sun, Mar 10, 2019 at 7:29 PM Driesprong, Fokko <fo...@driesprong.frl>
> wrote:
>
> Ps. Jarek, interesting idea. It shouldn't be too hard to make Airflow more
> > k8s native. You could package your dags within your container, and do a
> > rolling update. Add the DAGs as the last layer, and then point the DAGs
> > folder to the same location. The hard part here is that you need to
> > gracefuly restart the workers. Currently AFAIK the signals given to the
> pod
> > aren't respected. So when the scheduler/webserver/worker receives a
> > SIGTERM, it should stop the jobs nicely and then exit the container,
> before
> > k8s kills the container using a SIGKILL.  This will be challenging with
> the
> > workers, which they are potentially long-running. Maybe stop kicking off
> > new jobs, and let the old ones finish, will be good enough, but then we
> > need to increase the standard kill timeout substantially. Having this
> would
> > also enable the autoscaling of the workers
> >
>
>
> --
>
> Jarek Potiuk
> Polidea <https://www.polidea.com/> | Principal Software Engineer
>
> M: +48 660 796 129 <+48660796129>
> E: jarek.potiuk@polidea.com
>


-- 
*Greg Neiheisel* / CTO Astronomer.io

Pods not being TERMinated properly for K8s airflow

[Jarek Potiuk <Ja...@polidea.com>]

NOTE! I changed the subject to not pollute the AIP-12 thread

Fokko,

I think I know why TERM signals do not work in the current POD
implementation. I already experienced that several times  - dockerized app
not receiving TERM signal. The reason was always the same. It is not a bug
actually - it is expected behaviour in case your ENTRYPOINT is in the SHELL
form ("/binary arg1 arg2") or when you use shell script as ENTRYPOINT first
argument in [ "shell script", "arg" ] form.

In those cases "shell" process becomes the "0" init process. Unfortunately
shell process is not prepared to do all the stuff that proper init process
should be doing:

* Inherit orphaned child processes
* reap them
* Handle and propagate signals properly
* Wait until all subprocesses are terminated before terminating itself

What happens is that the TERM signal just kills the init "shell" process
but, then the signal does not reach any of its children and the container
continues to run. It's well known problem in docker world and there are a
number of solutions (including exec-ing from shell or - better - using a
dumb-init/tini in your ENTRYPOINT- very tiny "proper" init implementations
that do what they should do.

You can read more for example here: https://hynek.me/articles/docker-signals
.

J.

On Sun, Mar 10, 2019 at 7:29 PM Driesprong, Fokko <fo...@driesprong.frl>
wrote:

Ps. Jarek, interesting idea. It shouldn't be too hard to make Airflow more
> k8s native. You could package your dags within your container, and do a
> rolling update. Add the DAGs as the last layer, and then point the DAGs
> folder to the same location. The hard part here is that you need to
> gracefuly restart the workers. Currently AFAIK the signals given to the pod
> aren't respected. So when the scheduler/webserver/worker receives a
> SIGTERM, it should stop the jobs nicely and then exit the container, before
> k8s kills the container using a SIGKILL.  This will be challenging with the
> workers, which they are potentially long-running. Maybe stop kicking off
> new jobs, and let the old ones finish, will be good enough, but then we
> need to increase the standard kill timeout substantially. Having this would
> also enable the autoscaling of the workers
>


-- 

Jarek Potiuk
Polidea <https://www.polidea.com/> | Principal Software Engineer

M: +48 660 796 129 <+48660796129>
E: jarek.potiuk@polidea.com

Re: [DISCUSS] AIP-12 Persist DAG into DB

[Gabriel Silk <gs...@dropbox.com.INVALID>]

I'm really excited about this feature, and I'd love to be able to provide
feedback on the proposed design.

On Thu, Jul 18, 2019 at 10:21 AM Tao Feng <fe...@gmail.com> wrote:

> Thanks Ash. This will be huge!
>
> On Thu, Jul 18, 2019 at 4:00 AM Jarek Potiuk <Ja...@polidea.com>
> wrote:
>
> > Cool!
> >
> > On Thu, Jul 18, 2019 at 11:46 AM Ash Berlin-Taylor <as...@apache.org>
> wrote:
> >
> > > We didn't reach any conclusion on this yet but I agree, and this is the
> > > big task that we at Astronomer are going to work on next for Airflow.
> > >
> > > I've started chatting to a few of the other committers about this to
> get
> > a
> > > an idea of people's priorities, and have had a chat with Alex at Uber
> > about
> > > their experiences of making their internal fork of Airflow - Piper
> > > https://eng.uber.com/managing-data-workflows-at-scale/
> > >
> > > I'll create something in the wiki (probably not an AIP to start with)
> to
> > > collect the possible approaches and downsides/limitations.
> > >
> > > Watch this space.
> > >
> > > -ash
> > >
> > > > On 18 Jul 2019, at 07:05, Tao Feng <fe...@gmail.com> wrote:
> > > >
> > > > Do we reach any consensus on this topic /AIP? I think persisting DAG
> is
> > > > pretty important actually.
> > > >
> > > > -Tao
> > > >
> > > > On Tue, Mar 12, 2019 at 3:01 AM Kevin Yang <yr...@gmail.com>
> wrote:
> > > >
> > > >> Hi Fokko,
> > > >>
> > > >> As a large cluster maintainer, I’m not a big fan of large DAG files
> > > >> neither. But I’m not sure if I’ll consider this bad practice. We
> have
> > > some
> > > >> large frameworks, e.g. experimentation and machine learning, that
> are
> > > >> complex by nature and generate large number of DAGs from their
> > customer
> > > >> configs to get better flexibility. I consider them as advance use
> > cases
> > > of
> > > >> Airflow and open up a lot potentials for Airflow, unless we’ve
> > > previously
> > > >> set some boundaries around how complex DAG codes can be that I’m not
> > > aware
> > > >> of. About resulting in an unworkable situation, yes we are
> > experiencing
> > > >> pain from having such large DAG files, mainly on the webserver side,
> > but
> > > >> the system overall are running stable. We are actually hoping to
> > improve
> > > >> the situation by applying solutions like making webserver stateless.
> > It
> > > is
> > > >> ok that if the owners of large DAG files need to pay but we should
> try
> > > >> minimize the price—longer refresh interval, extra task running time,
> > but
> > > >> nothing too crazy.
> > > >>
> > > >>
> > > >> I think we’re aligned on storing info in DB as long as we can meet
> the
> > > >> requirements Dan mentioned earlier—we just need that balance
> decided,
> > so
> > > >> I’m gonna skip this part( out of all the requirements, No.1 seems to
> > be
> > > >> least clear, maybe we can expand on that). One thing about the
> > proposed
> > > >> idea is that we implicitly couple DagRun with DAG version, which at
> > the
> > > >> first glance make sense but imo not very ideal. I feel full
> versioning
> > > >> should track all changes instead of tracking changes only when we
> > create
> > > >> DagRun. E.g. my task failed and I merged new code to fix my task
> and I
> > > want
> > > >> to rerun it with the current code, with serialize DAG during DagRun
> > > >> creation time we won’t have the up to date snapshot—sure we can work
> > > around
> > > >> it by like always keep a current snapshot of DAG but this is kinda
> > messy
> > > >> and confusing. This is what popped up on the top of my head and w/o
> > full
> > > >> versioning we might have some other tricky cases, e.g. ur backfill
> > case.
> > > >> But I just gave a few thoughts into this and you might already have
> a
> > > >> complete story that will void my concerns.
> > > >>
> > > >>
> > > >> Cheers,
> > > >> Kevin Y
> > > >>
> > > >> On Sun, Mar 10, 2019 at 11:29 AM Driesprong, Fokko
> > <fokko@driesprong.frl
> > > >
> > > >> wrote:
> > > >>
> > > >>> Thanks Kevin for opening the discussion. I think it is important to
> > > have
> > > >> a
> > > >>> clear overview on how to approach the AIP.
> > > >>>
> > > >>> First of all, how many DAGs do we have that take 30s to parse? I
> > > consider
> > > >>> this bad practice, and this would also result in an unworkable
> > > situation
> > > >>> with the current setup of Airflow since it will take a lot of
> > resources
> > > >> on
> > > >>> the webserver/scheduler, and the whole system will become
> > > unresponsive. I
> > > >>> will be hard to cope with such DAGs in general.
> > > >>>
> > > >>> The idea from the AIP is to have the versioned version of the dag
> in
> > > the
> > > >>> DB, so in the end, you won't need to parse the whole thing every
> > time.
> > > >> Only
> > > >>> when you trigger a DAG, or when you want to see the current status
> of
> > > the
> > > >>> dag.
> > > >>>
> > > >>> Like stated earlier, I strongly feel we shouldn't serialize the
> DAGs
> > as
> > > >>> JSON(5) or pickles in general. For me, this is deferring the pain
> of
> > > >>> setting up a structure of the DAG object itself.
> > > >>> Having the DAG denormalized in the database will give us cleaner
> > > storage
> > > >> of
> > > >>> our DAG. We can, for example, enforce fields by making them not
> null,
> > > so
> > > >> we
> > > >>> know that is something is off at write time, instead of read.
> > > >> Furthermore,
> > > >>> we're missing logical types such as dates, which we efficiently can
> > > query
> > > >>> using the indices of the database.
> > > >>> Also, with all serialization formats, evolution isn't trivial.
> > Consider
> > > >> the
> > > >>> situations when:
> > > >>> - We're introducing a new field, and it might be null, therefore we
> > > need
> > > >> to
> > > >>> bake in all kinds of logic into the Airflow code, which you don't
> > want.
> > > >>> With proper migration scripts, you could prefill these fields, and
> > make
> > > >>> them not null.
> > > >>> - Changing the models, for example, you still can't change a
> > > string-type
> > > >>> into a integer with adding custom logic. In this case, the reviewer
> > > needs
> > > >>> to be extra careful that there are no breaking changes introduced.
> > > Right
> > > >>> now we're doing minimal forward- and backward compatibilitytesting.
> > > >>>
> > > >>> In the case we get too many migrations, we could also squash (some
> of
> > > >> them)
> > > >>> when preparing the release.
> > > >>>
> > > >>> Personally, I don't think the serialization is the issue here. As
> Max
> > > >>> already mentioned, it is the optimal balance of (de)normalization.
> > From
> > > >> the
> > > >>> user perspective, the serialization won't change much of the
> > behaviour
> > > of
> > > >>> Airflow.
> > > >>>
> > > >>> For me, instead of having `DAG.serialize()` and
> `DAG.deser(version)`
> > is
> > > >> not
> > > >>> the ideal approach. But it might be that we're on the same page
> :-) I
> > > >>> believe it should be something like `DagRun.find('fokkos_dag',
> > > >>> datetime(2018, 03, 01))` and construct the correct version of the
> > dag.
> > > >>> Since there is an uniqueness constrain on dag_id, datetime, this
> will
> > > >>> always return the same dag. You will get the versioned DagRun as it
> > ran
> > > >>> that time. Serializing the fields adn storing them in the database
> > > should
> > > >>> happen transparently when you update the DAG object. When you run a
> > > dag,
> > > >>> you'll parse the dag, and then run it. `Dag().create_dagrun(...)`,
> > this
> > > >>> will create a DagRun as the name suggests, if the version of the
> dag
> > > >> still
> > > >>> exists in the database, it will reuse that one, otherwise it will
> > > create
> > > >> a
> > > >>> new version of the DAG (with all the operators etc). In this sense
> > the
> > > >>> version of the DAGs should be done within the Dag(Run).
> > > >>>
> > > >>> The versioning will change the behavour from a user perspective.
> > Right
> > > >> now
> > > >>> we store only a single version. For example, the poor mans
> > backfilling
> > > >>> won't work anymore. This is clearing the state from past&future,
> up-
> > > and
> > > >>> downstream, and let it catch up again.
> > > >>> In this case, the old version of the DAG won't exists anymore, and
> > > >>> potentially there are tasks that aren't in the code anymore. In
> this
> > > case
> > > >>> we need to clear the version of the dag, and rerun it with the
> latest
> > > >>> version `DagRun.find('fokkos_dag', datetime(2018, 03,
> 01)).clear()`.
> > > How
> > > >> we
> > > >>> are going to do clear's downstram in the middle of the dag, that is
> > > >>> something I still have to figure out. Because potentially there are
> > > tasks
> > > >>> that can't be rerun because the underlying Python code has changed,
> > > both
> > > >> on
> > > >>> user level as on Airflow level. It will be impossible to get these
> > > >> features
> > > >>> pure in that sense.
> > > >>> I would not suggest adding a new status in here, indicating that
> the
> > > task
> > > >>> can't be rerun since it isn't part of the DAG anymore. We have to
> > find
> > > >> the
> > > >>> balance here in adding complexity (also to the scheduler) and
> > features
> > > >> that
> > > >>> we need to introduce to help the user.
> > > >>>
> > > >>> Cheers, Fokko
> > > >>>
> > > >>> Ps. Jarek, interesting idea. It shouldn't be too hard to make
> Airflow
> > > >> more
> > > >>> k8s native. You could package your dags within your container, and
> > do a
> > > >>> rolling update. Add the DAGs as the last layer, and then point the
> > DAGs
> > > >>> folder to the same location. The hard part here is that you need to
> > > >>> gracefuly restart the workers. Currently AFAIK the signals given to
> > the
> > > >> pod
> > > >>> aren't respected. So when the scheduler/webserver/worker receives a
> > > >>> SIGTERM, it should stop the jobs nicely and then exit the
> container,
> > > >> before
> > > >>> k8s kills the container using a SIGKILL.  This will be challenging
> > with
> > > >> the
> > > >>> workers, which they are potentially long-running. Maybe stop
> kicking
> > > off
> > > >>> new jobs, and let the old ones finish, will be good enough, but
> then
> > we
> > > >>> need to increase the standard kill timeout substantially. Having
> this
> > > >> would
> > > >>> also enable the autoscaling of the workers.
> > > >>>
> > > >>>
> > > >>>
> > > >>> Op za 9 mrt. 2019 om 19:07 schreef Maxime Beauchemin <
> > > >>> maximebeauchemin@gmail.com>:
> > > >>>
> > > >>>> I want to raise the question of the amount of normalization we
> want
> > to
> > > >>> use
> > > >>>> here as it seems the to be an area that needs more attention.
> > > >>>>
> > > >>>> The SIP suggest having DAG blobs, task blobs and edges (call it
> the
> > > >>>> fairly-normalized). I also like the idea of all-encompassing (call
> > it
> > > >>>> very-denormalized) DAG blobs as it seems easier to manage in terms
> > of
> > > >>>> versioning. The question here is whether we go with one of these
> > > method
> > > >>>> exclusively, something in-between or even a hybrid approach
> > (redundant
> > > >>>> blobs that use different level of normalization).
> > > >>>>
> > > >>>> It's nice and simple to just push or pull DAG atomic objects with
> a
> > > >>> version
> > > >>>> stamp on it. It's clearly simpler than dealing with 3 versioned
> > tables
> > > >>>> (dag, tasks, edges). There are a lot of pros/cons, and they become
> > > more
> > > >>>> apparent with the perspective of very large DAGs. If the web
> server
> > is
> > > >>>> building a "task details page", using the "fairly-normalized"
> model,
> > > it
> > > >>> can
> > > >>>> just pull what it needs instead of pulling the large DAG blob.
> > > >> Similarly,
> > > >>>> if building a sub-tree view (a subset of the DAG), perhaps it can
> > only
> > > >>>> retrieve what it needs. But if you need the whole DAG (say for the
> > > >>>> scheduler use case) then you're dealing with more complex SQL/ORM
> > > >>>> operations (joins hopefully, or multiple db round trips)
> > > >>>>
> > > >>>> Now maybe the right approach is more something like 2 tables: DAG
> > and
> > > >>>> task_details, where edges keys are denormalized into DAG (arguably
> > > >>> that's a
> > > >>>> few KBs at most, even for large DAGs), and maybe the DAG object
> has
> > > >> most
> > > >>> of
> > > >>>> the high level task metadata information (operator, name,
> > baseoperator
> > > >>> key
> > > >>>> attrs), and task_details has the big blobs (SQL code). This is
> > > >> probably a
> > > >>>> nice compromise, the question becomes "how much task-level detail
> do
> > > we
> > > >>>> store in the DAG-centric blog?", probably not much to keep the DAG
> > > >>> objects
> > > >>>> as small as possible. The main downside here is that you cannot
> have
> > > >> the
> > > >>>> database join and have to do 2 round trips to reconstruct a DAG
> > object
> > > >>>> (fetch the DAG, parse the object to get the list of tasks, and
> then
> > > run
> > > >>>> another db query to get those task details).
> > > >>>>
> > > >>>> To resume, I'd qualify the more normalized approach as the most
> > > proper,
> > > >>> but
> > > >>>> also the more complex. It'll shine in specific cases around large
> > > DAGs.
> > > >>> If
> > > >>>> we have the proper abstractions (methods like DAG.serialize(),
> > > >>>> DAG.deser(version)) then I guess that's not an issue.
> > > >>>>
> > > >>>> Max
> > > >>>>
> > > >>>> On Fri, Mar 8, 2019 at 5:21 PM Kevin Yang <yr...@gmail.com>
> > wrote:
> > > >>>>
> > > >>>>> Hi Julian, I'm definitely aligned with you guys on making the
> > > >> webserver
> > > >>>>> independent of DAG parsing, just the end goal to me would be to
> > > >> build a
> > > >>>>> complete story around serializing DAG--and move with the story in
> > > >>> mind. I
> > > >>>>> feel like you guys may already have a list of dynamic features we
> > > >> need
> > > >>> to
> > > >>>>> deprecate/change, if that is the case feel free to open the
> > > >> discussion
> > > >>> on
> > > >>>>> what we do to them with DAG serialization.
> > > >>>>>
> > > >>>>> Julian, Ash, Dan, on 2nd thought I do agree that if we can meet
> the
> > > >>>>> requirements Dan mentioned, it would be nice to have them stored
> in
> > > >> the
> > > >>>> DB.
> > > >>>>> Some combined solutions like having a column of serialized graph
> in
> > > >> the
> > > >>>>> serialized dag table can potentially meet all requirements. What
> > > >> format
> > > >>>> we
> > > >>>>> end up using to represent DAG between components is now less
> > > >> important
> > > >>>>> IMO--fine to refactor those endpoints only need DagModel to use
> > only
> > > >>>>> DagModel, easy to do a batch replacement if we decide otherwise
> > > >> later.
> > > >>>> More
> > > >>>>> important is to define this source of truth for serialized DAG.
> > > >>>>>
> > > >>>>> Ash, ty for the email list, I'll tune my filters accordingly :D
> I'm
> > > >>>> leaning
> > > >>>>> towards having a separate process for the parser so we got no
> > > >> scheduler
> > > >>>>> dependency etc for this parser but we can discuss this in another
> > > >>> thread.
> > > >>>>>
> > > >>>>> On Fri, Mar 8, 2019 at 8:57 AM Dan Davydov
> > > >>> <ddavydov@twitter.com.invalid
> > > >>>>>
> > > >>>>> wrote:
> > > >>>>>
> > > >>>>>>>
> > > >>>>>>> Personally I don’t understand why people are pushing for a
> > > >>> JSON-based
> > > >>>>> DAG
> > > >>>>>>> representation
> > > >>>>>>
> > > >>>>>> It sounds like you agree that DAGs should be serialized (just in
> > > >> the
> > > >>> DB
> > > >>>>>> instead of JSON), so will only address why JSON is better than
> > > >> MySQL
> > > >>>> (AKA
> > > >>>>>> serializing at the DAG level vs the task level) as far as I can
> > > >> see,
> > > >>>> and
> > > >>>>>> not why we need serialization. If you zoom out and look at all
> the
> > > >>> use
> > > >>>>>> cases of serialized DAGs, e.g. having the scheduler use them
> > > >> instead
> > > >>> of
> > > >>>>>> parsing DAGs directly, then it becomes clear that we need all
> > > >>>> appropriate
> > > >>>>>> metadata in these DAGs, (operator params, DAG properties, etc),
> in
> > > >>>> which
> > > >>>>>> case it's not clear how it will fit nicely into a DB table
> (unless
> > > >>> you
> > > >>>>>> wanted to do something like (parent_task_id, task_id,
> > task_params),
> > > >>>> also
> > > >>>>>> keep in mind that we will need to store different versions of
> each
> > > >>> DAG
> > > >>>> in
> > > >>>>>> the future so that we can ensure consistency in a dagrun, i.e.
> > each
> > > >>>> task
> > > >>>>> in
> > > >>>>>> a dagrun uses the same version of a DAG.
> > > >>>>>>
> > > >>>>>> I think some of our requirements should be:
> > > >>>>>> 1. The data model will lead to acceptable performance in all of
> > its
> > > >>>>>> consumers (scheduler, webserver, workers), i.e. no n+1 access
> > > >>> patterns
> > > >>>>> (my
> > > >>>>>> biggest concern about serializing at task level as you propose
> vs
> > > >> at
> > > >>>> DAG
> > > >>>>>> level)
> > > >>>>>> 2. We can have versioning of serialized DAGs
> > > >>>>>> 3. The ability to separate DAGs into their own data store (e.g.
> no
> > > >>>>> reliance
> > > >>>>>> on joins between the new table and the old one)
> > > >>>>>> 4. One source of truth/serialized representation for DAGs
> > > >> (currently
> > > >>> we
> > > >>>>>> have SimpleDAG)
> > > >>>>>>
> > > >>>>>> If we can full-fill all of these requirements and serialize at
> the
> > > >>> task
> > > >>>>>> level rather than the DAG level in the DB, then I agree that
> > > >> probably
> > > >>>>> makes
> > > >>>>>> more sense.
> > > >>>>>>
> > > >>>>>>
> > > >>>>>>> In the proposed PR’s we (Peter, Bas and me) aim to avoid
> > > >> re-parsing
> > > >>>> DAG
> > > >>>>>>> files by querying all the required information from the
> database.
> > > >>> In
> > > >>>>> one
> > > >>>>>> or
> > > >>>>>>> two cases this may however not be possible, in which case we
> > > >> might
> > > >>>>> either
> > > >>>>>>> have to fall back on the DAG file or add the missing
> information
> > > >>> into
> > > >>>>> the
> > > >>>>>>> database. We can tackle these problems as we encounter them.
> > > >>>>>>
> > > >>>>>> I think you would have the support of many of committers in
> > > >> removing
> > > >>>> any
> > > >>>>>> use-cases that stand in the way of full serialization, that
> being
> > > >>> said
> > > >>>> if
> > > >>>>>> we need to remove features we need to do this carefully and
> > > >>>> thoughtfully,
> > > >>>>>> and ideally with proposed alternatives/work-arounds to cover the
> > > >>>>> removals.
> > > >>>>>>
> > > >>>>>> The counter argument: this PR removes the need for the confusing
> > > >>>>> "Refresh"
> > > >>>>>>> button from the UI, and in general you only pay the cost for
> the
> > > >>>>>> expensive
> > > >>>>>>> DAGs when you ask about them. (I don't know what/when we call
> the
> > > >>>>>>> /pickle_info endpoint of the top of my head)
> > > >>>>>>
> > > >>>>>> Probably worth splitting out into a separate thread, but I'm
> > > >> actually
> > > >>>> not
> > > >>>>>> sure the refresh button does anything, I think we should double
> > > >>>> check...
> > > >>>>> I
> > > >>>>>> think about 2 years ago there was a commit made that made
> gunicorn
> > > >>>>>> webservers automatically rotate underneath flask (each one would
> > > >>>> reparse
> > > >>>>>> the DAGbag). Even if it works we should probably remove it since
> > > >> the
> > > >>>>>> webserver refresh interval is pretty fast, and it just causes
> > > >>> confusion
> > > >>>>> to
> > > >>>>>> users and implies that the DAGs are not refreshed automatically.
> > > >>>>>>
> > > >>>>>> Do you mean https://json5.org/ or is this a typo? That might be
> > > >> okay
> > > >>>>> for a
> > > >>>>>>> nicer user front end, but the "canonical" version stored in the
> > > >> DB
> > > >>>>> should
> > > >>>>>>> be something "plainer" like just JSON.
> > > >>>>>>
> > > >>>>>> I think he got this from my reply, and it was just an example,
> but
> > > >>> you
> > > >>>>> are
> > > >>>>>> right, I agree JSON would be better than JSON5.
> > > >>>>>>
> > > >>>>>> On Fri, Mar 8, 2019 at 8:53 AM Ash Berlin-Taylor <
> ash@apache.org>
> > > >>>> wrote:
> > > >>>>>>
> > > >>>>>>> Comments inline.
> > > >>>>>>>
> > > >>>>>>>> On 8 Mar 2019, at 11:28, Kevin Yang <yr...@gmail.com>
> wrote:
> > > >>>>>>>>
> > > >>>>>>>> Hi all,
> > > >>>>>>>> When I was preparing some work related to this AIP I found
> > > >>>> something
> > > >>>>>>> very concerning. I noticed this JIRA ticket <
> > > >>>>>>> https://issues.apache.org/jira/browse/AIRFLOW-3562> is trying
> to
> > > >>>>> remove
> > > >>>>>>> the dependency of dagbag from webserver, which is awesome--we
> > > >>> wanted
> > > >>>>>> badly
> > > >>>>>>> but never got to start work on. However when I looked at some
> > > >>>> subtasks
> > > >>>>> of
> > > >>>>>>> it, which try to remove dagbag dependency from each endpoint, I
> > > >>> found
> > > >>>>> the
> > > >>>>>>> way we remove the dependency of dagbag is not very ideal. For
> > > >>> example
> > > >>>>>> this
> > > >>>>>>> PR <https://github.com/apache/airflow/pull/4867/files> will
> > > >>> require
> > > >>>> us
> > > >>>>>> to
> > > >>>>>>> parse the dag file each time we hit the endpoint.
> > > >>>>>>>
> > > >>>>>>> The counter argument: this PR removes the need for the
> confusing
> > > >>>>>> "Refresh"
> > > >>>>>>> button from the UI, and in general you only pay the cost for
> the
> > > >>>>>> expensive
> > > >>>>>>> DAGs when you ask about them. (I don't know what/when we call
> the
> > > >>>>>>> /pickle_info endpoint of the top of my head)
> > > >>>>>>>
> > > >>>>>>> This end point may be one to hold off on (as it can ask for
> > > >>> multiple
> > > >>>>>> dags)
> > > >>>>>>> but there are some that def don't need a full dag bag or to
> even
> > > >>>> parse
> > > >>>>>> the
> > > >>>>>>> dag file, the current DAG model has enough info.
> > > >>>>>>>
> > > >>>>>>>>
> > > >>>>>>>>
> > > >>>>>>>> If we go down this path, we indeed can get rid of the dagbag
> > > >>>>> dependency
> > > >>>>>>> easily, but we will have to 1. increase the DB load( not too
> > > >>>> concerning
> > > >>>>>> at
> > > >>>>>>> the moment ), 2. wait the DAG file to be parsed before getting
> > > >> the
> > > >>>> page
> > > >>>>>>> back, potentially multiple times. DAG file can sometimes take
> > > >>> quite a
> > > >>>>>> while
> > > >>>>>>> to parse, e.g. we have some framework DAG files generating
> large
> > > >>>> number
> > > >>>>>> of
> > > >>>>>>> DAGs from some static config files or even jupyter notebooks
> and
> > > >>> they
> > > >>>>> can
> > > >>>>>>> take 30+ seconds to parse. Yes we don't like large DAG files
> but
> > > >>>> people
> > > >>>>>> do
> > > >>>>>>> see the beauty of code as config and sometimes heavily
> > > >>> abuseleverage
> > > >>>>> it.
> > > >>>>>>> Assuming all users have the same nice small python file that
> can
> > > >> be
> > > >>>>>> parsed
> > > >>>>>>> fast, I'm still a bit worried about this approach. Continuing
> on
> > > >>> this
> > > >>>>>> path
> > > >>>>>>> means we've chosen DagModel to be the serialized representation
> > > >> of
> > > >>>> DAG
> > > >>>>>> and
> > > >>>>>>> DB columns to hold different properties--it can be one
> candidate
> > > >>> but
> > > >>>> I
> > > >>>>>>> don't know if we should settle on that now. I would personally
> > > >>>> prefer a
> > > >>>>>>> more compact, e.g. JSON5, and easy to scale representation(
> such
> > > >>> that
> > > >>>>>>> serializing new fields != DB upgrade).
> > > >>>>>>>
> > > >>>>>>> Do you mean https://json5.org/ or is this a typo? That might
> be
> > > >>> okay
> > > >>>>> for
> > > >>>>>>> a nicer user front end, but the "canonical" version stored in
> the
> > > >>> DB
> > > >>>>>> should
> > > >>>>>>> be something "plainer" like just JSON.
> > > >>>>>>>
> > > >>>>>>> I'm not sure that "serializing new fields != DB upgrade" is
> that
> > > >>> big
> > > >>>>> of a
> > > >>>>>>> concern, as we don't add fields that often. One possible way of
> > > >>>> dealing
> > > >>>>>>> with it if we do is to have a hybrid approach - a few distinct
> > > >>>> columns,
> > > >>>>>> but
> > > >>>>>>> then a JSON blob. (and if we were only to support postgres we
> > > >> could
> > > >>>>> just
> > > >>>>>>> use JSONb. But I think our friends at Google may object ;) )
> > > >>>>>>>
> > > >>>>>>> Adding a new column in a DB migration with a default NULL
> > > >> shouldn't
> > > >>>> be
> > > >>>>> an
> > > >>>>>>> expensive operation, or difficult to achieve.
> > > >>>>>>>
> > > >>>>>>>
> > > >>>>>>>>
> > > >>>>>>>> In my imagination we would have to collect the list of dynamic
> > > >>>>> features
> > > >>>>>>> depending on unserializable fields of a DAG and start a
> > > >>>> discussion/vote
> > > >>>>>> on
> > > >>>>>>> dropping support of them( I'm working on this but if anyone has
> > > >>>> already
> > > >>>>>>> done so please take over), decide on the serialized
> > > >> representation
> > > >>>> of a
> > > >>>>>> DAG
> > > >>>>>>> and then replace dagbag with it in webserver. Per previous
> > > >>> discussion
> > > >>>>> and
> > > >>>>>>> some offline discussions with Dan, one future of DAG
> > > >> serialization
> > > >>>>> that I
> > > >>>>>>> like would look similar to this:
> > > >>>>>>>>
> > > >>>>>>>
> > > >>>>>>>> https://imgur.com/ncqqQgc
> > > >>>>>>>
> > > >>>>>>> Something I've thought about before for other things was to
> embed
> > > >>> an
> > > >>>>> API
> > > >>>>>>> server _into_ the scheduler - this would be useful for k8s
> > > >>>>> healthchecks,
> > > >>>>>>> native Prometheus metrics without needed statsd bridge, and
> could
> > > >>>> have
> > > >>>>>>> endpoints to get information such as this directly.
> > > >>>>>>>
> > > >>>>>>> I was thinking it would be _in_ the scheduler process using
> > > >> either
> > > >>>>>> threads
> > > >>>>>>> (ick. Python's still got a GIL doesn't it?) or using
> > > >> async/twisted
> > > >>>> etc.
> > > >>>>>>> (not a side-car process like we have with the logs webserver
> for
> > > >>>>> `airflow
> > > >>>>>>> worker`).
> > > >>>>>>>
> > > >>>>>>> (This is possibly an unrelated discussion, but might be worth
> > > >>> talking
> > > >>>>>>> about?)
> > > >>>>>>>
> > > >>>>>>>> We can still discuss/vote which approach we want to take but I
> > > >>>> don't
> > > >>>>>>> want the door to above design to be shut right now or we have
> to
> > > >>>> spend
> > > >>>>> a
> > > >>>>>>> lot effort switch path later.
> > > >>>>>>>>
> > > >>>>>>>> Bas and Peter, I'm very sorry to extend the discussion but I
> do
> > > >>>> think
> > > >>>>>>> this is tightly related to the AIP and PRs behind it. And my
> > > >>> sincere
> > > >>>>>>> apology for bringing this up so late( I only pull the open PR
> > > >> list
> > > >>>>>>> occasionally, if there's a way to subscribe to new PR event I'd
> > > >>> love
> > > >>>> to
> > > >>>>>>> know how).
> > > >>>>>>>
> > > >>>>>>> It's noisy, but you can subscribe to
> commits@airflow.apache.org
> > > >>> (but
> > > >>>>> be
> > > >>>>>>> warned, this also includes all Jira tickets, edits of every
> > > >> comment
> > > >>>> on
> > > >>>>>>> github etc.).
> > > >>>>>>>
> > > >>>>>>>
> > > >>>>>>>>
> > > >>>>>>>> Cheers,
> > > >>>>>>>> Kevin Y
> > > >>>>>>>>
> > > >>>>>>>> On Thu, Feb 28, 2019 at 1:36 PM Peter van t Hof <
> > > >>>>> pjrvanthof@gmail.com
> > > >>>>>>> <ma...@gmail.com>> wrote:
> > > >>>>>>>> Hi all,
> > > >>>>>>>>
> > > >>>>>>>> Just some comments one the point Bolke dit give in relation of
> > > >> my
> > > >>>> PR.
> > > >>>>>>>>
> > > >>>>>>>> At first, the main focus is: making the webserver stateless.
> > > >>>>>>>>
> > > >>>>>>>>> 1) Make the webserver stateless: needs the graph of the
> > > >>> *current*
> > > >>>>> dag
> > > >>>>>>>>
> > > >>>>>>>> This is the main goal but for this a lot more PR’s will be
> > > >> coming
> > > >>>>> once
> > > >>>>>>> my current is merged. For edges and graph view this is covered
> in
> > > >>> my
> > > >>>> PR
> > > >>>>>>> already.
> > > >>>>>>>>
> > > >>>>>>>>> 2) Version dags: for consistency mainly and not requiring
> > > >>> parsing
> > > >>>>> of
> > > >>>>>>> the
> > > >>>>>>>>> dag on every loop
> > > >>>>>>>>
> > > >>>>>>>> In my PR the historical graphs will be stored for each DagRun.
> > > >>> This
> > > >>>>>>> means that you can see if an older DagRun was the same graph
> > > >>>> structure,
> > > >>>>>>> even if some tasks does not exists anymore in the current
> graph.
> > > >>>>>> Especially
> > > >>>>>>> for dynamic DAG’s this is very useful.
> > > >>>>>>>>
> > > >>>>>>>>> 3) Make the scheduler not require DAG files. This could be
> > > >> done
> > > >>>> if
> > > >>>>>> the
> > > >>>>>>>>> edges contain all information when to trigger the next task.
> > > >> We
> > > >>>> can
> > > >>>>>>> then
> > > >>>>>>>>> have event driven dag parsing outside of the scheduler loop,
> > > >>> ie.
> > > >>>> by
> > > >>>>>> the
> > > >>>>>>>>> cli. Storage can also be somewhere else (git, artifactory,
> > > >>>>>> filesystem,
> > > >>>>>>>>> whatever).
> > > >>>>>>>>
> > > >>>>>>>> The scheduler is almost untouched in this PR. The only thing
> > > >> that
> > > >>>> is
> > > >>>>>>> added is that this edges are saved to the database but the
> > > >>> scheduling
> > > >>>>>>> itself din’t change. The scheduler depends now still on the DAG
> > > >>>> object.
> > > >>>>>>>>
> > > >>>>>>>>> 4) Fully serialise the dag so it becomes transferable to
> > > >>> workers
> > > >>>>>>>>
> > > >>>>>>>> It nice to see that people has a lot of idea’s about this. But
> > > >> as
> > > >>>>> Fokko
> > > >>>>>>> already mentioned this is out of scope for the issue what we
> are
> > > >>>> trying
> > > >>>>>> to
> > > >>>>>>> solve. I also have some idea’s about this but I like to limit
> > > >> this
> > > >>>>> PR/AIP
> > > >>>>>>> to the webserver.
> > > >>>>>>>>
> > > >>>>>>>> For now my PR does solve 1 and 2 and the rest of the behaviour
> > > >>>> (like
> > > >>>>>>> scheduling) is untouched.
> > > >>>>>>>>
> > > >>>>>>>> Gr,
> > > >>>>>>>> Peter
> > > >>>>>>>>
> > > >>>>>>>
> > > >>>>>>>
> > > >>>>>>
> > > >>>>>
> > > >>>>
> > > >>>
> > > >>
> > >
> > >
> >
> > --
> >
> > Jarek Potiuk
> > Polidea <https://www.polidea.com/> | Principal Software Engineer
> >
> > M: +48 660 796 129 <+48660796129>
> > [image: Polidea] <https://www.polidea.com/>
> >
>

Re: [DISCUSS] AIP-12 Persist DAG into DB

[Tao Feng <fe...@gmail.com>]

Thanks Ash. This will be huge!

On Thu, Jul 18, 2019 at 4:00 AM Jarek Potiuk <Ja...@polidea.com>
wrote:

> Cool!
>
> On Thu, Jul 18, 2019 at 11:46 AM Ash Berlin-Taylor <as...@apache.org> wrote:
>
> > We didn't reach any conclusion on this yet but I agree, and this is the
> > big task that we at Astronomer are going to work on next for Airflow.
> >
> > I've started chatting to a few of the other committers about this to get
> a
> > an idea of people's priorities, and have had a chat with Alex at Uber
> about
> > their experiences of making their internal fork of Airflow - Piper
> > https://eng.uber.com/managing-data-workflows-at-scale/
> >
> > I'll create something in the wiki (probably not an AIP to start with) to
> > collect the possible approaches and downsides/limitations.
> >
> > Watch this space.
> >
> > -ash
> >
> > > On 18 Jul 2019, at 07:05, Tao Feng <fe...@gmail.com> wrote:
> > >
> > > Do we reach any consensus on this topic /AIP? I think persisting DAG is
> > > pretty important actually.
> > >
> > > -Tao
> > >
> > > On Tue, Mar 12, 2019 at 3:01 AM Kevin Yang <yr...@gmail.com> wrote:
> > >
> > >> Hi Fokko,
> > >>
> > >> As a large cluster maintainer, I’m not a big fan of large DAG files
> > >> neither. But I’m not sure if I’ll consider this bad practice. We have
> > some
> > >> large frameworks, e.g. experimentation and machine learning, that are
> > >> complex by nature and generate large number of DAGs from their
> customer
> > >> configs to get better flexibility. I consider them as advance use
> cases
> > of
> > >> Airflow and open up a lot potentials for Airflow, unless we’ve
> > previously
> > >> set some boundaries around how complex DAG codes can be that I’m not
> > aware
> > >> of. About resulting in an unworkable situation, yes we are
> experiencing
> > >> pain from having such large DAG files, mainly on the webserver side,
> but
> > >> the system overall are running stable. We are actually hoping to
> improve
> > >> the situation by applying solutions like making webserver stateless.
> It
> > is
> > >> ok that if the owners of large DAG files need to pay but we should try
> > >> minimize the price—longer refresh interval, extra task running time,
> but
> > >> nothing too crazy.
> > >>
> > >>
> > >> I think we’re aligned on storing info in DB as long as we can meet the
> > >> requirements Dan mentioned earlier—we just need that balance decided,
> so
> > >> I’m gonna skip this part( out of all the requirements, No.1 seems to
> be
> > >> least clear, maybe we can expand on that). One thing about the
> proposed
> > >> idea is that we implicitly couple DagRun with DAG version, which at
> the
> > >> first glance make sense but imo not very ideal. I feel full versioning
> > >> should track all changes instead of tracking changes only when we
> create
> > >> DagRun. E.g. my task failed and I merged new code to fix my task and I
> > want
> > >> to rerun it with the current code, with serialize DAG during DagRun
> > >> creation time we won’t have the up to date snapshot—sure we can work
> > around
> > >> it by like always keep a current snapshot of DAG but this is kinda
> messy
> > >> and confusing. This is what popped up on the top of my head and w/o
> full
> > >> versioning we might have some other tricky cases, e.g. ur backfill
> case.
> > >> But I just gave a few thoughts into this and you might already have a
> > >> complete story that will void my concerns.
> > >>
> > >>
> > >> Cheers,
> > >> Kevin Y
> > >>
> > >> On Sun, Mar 10, 2019 at 11:29 AM Driesprong, Fokko
> <fokko@driesprong.frl
> > >
> > >> wrote:
> > >>
> > >>> Thanks Kevin for opening the discussion. I think it is important to
> > have
> > >> a
> > >>> clear overview on how to approach the AIP.
> > >>>
> > >>> First of all, how many DAGs do we have that take 30s to parse? I
> > consider
> > >>> this bad practice, and this would also result in an unworkable
> > situation
> > >>> with the current setup of Airflow since it will take a lot of
> resources
> > >> on
> > >>> the webserver/scheduler, and the whole system will become
> > unresponsive. I
> > >>> will be hard to cope with such DAGs in general.
> > >>>
> > >>> The idea from the AIP is to have the versioned version of the dag in
> > the
> > >>> DB, so in the end, you won't need to parse the whole thing every
> time.
> > >> Only
> > >>> when you trigger a DAG, or when you want to see the current status of
> > the
> > >>> dag.
> > >>>
> > >>> Like stated earlier, I strongly feel we shouldn't serialize the DAGs
> as
> > >>> JSON(5) or pickles in general. For me, this is deferring the pain of
> > >>> setting up a structure of the DAG object itself.
> > >>> Having the DAG denormalized in the database will give us cleaner
> > storage
> > >> of
> > >>> our DAG. We can, for example, enforce fields by making them not null,
> > so
> > >> we
> > >>> know that is something is off at write time, instead of read.
> > >> Furthermore,
> > >>> we're missing logical types such as dates, which we efficiently can
> > query
> > >>> using the indices of the database.
> > >>> Also, with all serialization formats, evolution isn't trivial.
> Consider
> > >> the
> > >>> situations when:
> > >>> - We're introducing a new field, and it might be null, therefore we
> > need
> > >> to
> > >>> bake in all kinds of logic into the Airflow code, which you don't
> want.
> > >>> With proper migration scripts, you could prefill these fields, and
> make
> > >>> them not null.
> > >>> - Changing the models, for example, you still can't change a
> > string-type
> > >>> into a integer with adding custom logic. In this case, the reviewer
> > needs
> > >>> to be extra careful that there are no breaking changes introduced.
> > Right
> > >>> now we're doing minimal forward- and backward compatibilitytesting.
> > >>>
> > >>> In the case we get too many migrations, we could also squash (some of
> > >> them)
> > >>> when preparing the release.
> > >>>
> > >>> Personally, I don't think the serialization is the issue here. As Max
> > >>> already mentioned, it is the optimal balance of (de)normalization.
> From
> > >> the
> > >>> user perspective, the serialization won't change much of the
> behaviour
> > of
> > >>> Airflow.
> > >>>
> > >>> For me, instead of having `DAG.serialize()` and `DAG.deser(version)`
> is
> > >> not
> > >>> the ideal approach. But it might be that we're on the same page :-) I
> > >>> believe it should be something like `DagRun.find('fokkos_dag',
> > >>> datetime(2018, 03, 01))` and construct the correct version of the
> dag.
> > >>> Since there is an uniqueness constrain on dag_id, datetime, this will
> > >>> always return the same dag. You will get the versioned DagRun as it
> ran
> > >>> that time. Serializing the fields adn storing them in the database
> > should
> > >>> happen transparently when you update the DAG object. When you run a
> > dag,
> > >>> you'll parse the dag, and then run it. `Dag().create_dagrun(...)`,
> this
> > >>> will create a DagRun as the name suggests, if the version of the dag
> > >> still
> > >>> exists in the database, it will reuse that one, otherwise it will
> > create
> > >> a
> > >>> new version of the DAG (with all the operators etc). In this sense
> the
> > >>> version of the DAGs should be done within the Dag(Run).
> > >>>
> > >>> The versioning will change the behavour from a user perspective.
> Right
> > >> now
> > >>> we store only a single version. For example, the poor mans
> backfilling
> > >>> won't work anymore. This is clearing the state from past&future, up-
> > and
> > >>> downstream, and let it catch up again.
> > >>> In this case, the old version of the DAG won't exists anymore, and
> > >>> potentially there are tasks that aren't in the code anymore. In this
> > case
> > >>> we need to clear the version of the dag, and rerun it with the latest
> > >>> version `DagRun.find('fokkos_dag', datetime(2018, 03, 01)).clear()`.
> > How
> > >> we
> > >>> are going to do clear's downstram in the middle of the dag, that is
> > >>> something I still have to figure out. Because potentially there are
> > tasks
> > >>> that can't be rerun because the underlying Python code has changed,
> > both
> > >> on
> > >>> user level as on Airflow level. It will be impossible to get these
> > >> features
> > >>> pure in that sense.
> > >>> I would not suggest adding a new status in here, indicating that the
> > task
> > >>> can't be rerun since it isn't part of the DAG anymore. We have to
> find
> > >> the
> > >>> balance here in adding complexity (also to the scheduler) and
> features
> > >> that
> > >>> we need to introduce to help the user.
> > >>>
> > >>> Cheers, Fokko
> > >>>
> > >>> Ps. Jarek, interesting idea. It shouldn't be too hard to make Airflow
> > >> more
> > >>> k8s native. You could package your dags within your container, and
> do a
> > >>> rolling update. Add the DAGs as the last layer, and then point the
> DAGs
> > >>> folder to the same location. The hard part here is that you need to
> > >>> gracefuly restart the workers. Currently AFAIK the signals given to
> the
> > >> pod
> > >>> aren't respected. So when the scheduler/webserver/worker receives a
> > >>> SIGTERM, it should stop the jobs nicely and then exit the container,
> > >> before
> > >>> k8s kills the container using a SIGKILL.  This will be challenging
> with
> > >> the
> > >>> workers, which they are potentially long-running. Maybe stop kicking
> > off
> > >>> new jobs, and let the old ones finish, will be good enough, but then
> we
> > >>> need to increase the standard kill timeout substantially. Having this
> > >> would
> > >>> also enable the autoscaling of the workers.
> > >>>
> > >>>
> > >>>
> > >>> Op za 9 mrt. 2019 om 19:07 schreef Maxime Beauchemin <
> > >>> maximebeauchemin@gmail.com>:
> > >>>
> > >>>> I want to raise the question of the amount of normalization we want
> to
> > >>> use
> > >>>> here as it seems the to be an area that needs more attention.
> > >>>>
> > >>>> The SIP suggest having DAG blobs, task blobs and edges (call it the
> > >>>> fairly-normalized). I also like the idea of all-encompassing (call
> it
> > >>>> very-denormalized) DAG blobs as it seems easier to manage in terms
> of
> > >>>> versioning. The question here is whether we go with one of these
> > method
> > >>>> exclusively, something in-between or even a hybrid approach
> (redundant
> > >>>> blobs that use different level of normalization).
> > >>>>
> > >>>> It's nice and simple to just push or pull DAG atomic objects with a
> > >>> version
> > >>>> stamp on it. It's clearly simpler than dealing with 3 versioned
> tables
> > >>>> (dag, tasks, edges). There are a lot of pros/cons, and they become
> > more
> > >>>> apparent with the perspective of very large DAGs. If the web server
> is
> > >>>> building a "task details page", using the "fairly-normalized" model,
> > it
> > >>> can
> > >>>> just pull what it needs instead of pulling the large DAG blob.
> > >> Similarly,
> > >>>> if building a sub-tree view (a subset of the DAG), perhaps it can
> only
> > >>>> retrieve what it needs. But if you need the whole DAG (say for the
> > >>>> scheduler use case) then you're dealing with more complex SQL/ORM
> > >>>> operations (joins hopefully, or multiple db round trips)
> > >>>>
> > >>>> Now maybe the right approach is more something like 2 tables: DAG
> and
> > >>>> task_details, where edges keys are denormalized into DAG (arguably
> > >>> that's a
> > >>>> few KBs at most, even for large DAGs), and maybe the DAG object has
> > >> most
> > >>> of
> > >>>> the high level task metadata information (operator, name,
> baseoperator
> > >>> key
> > >>>> attrs), and task_details has the big blobs (SQL code). This is
> > >> probably a
> > >>>> nice compromise, the question becomes "how much task-level detail do
> > we
> > >>>> store in the DAG-centric blog?", probably not much to keep the DAG
> > >>> objects
> > >>>> as small as possible. The main downside here is that you cannot have
> > >> the
> > >>>> database join and have to do 2 round trips to reconstruct a DAG
> object
> > >>>> (fetch the DAG, parse the object to get the list of tasks, and then
> > run
> > >>>> another db query to get those task details).
> > >>>>
> > >>>> To resume, I'd qualify the more normalized approach as the most
> > proper,
> > >>> but
> > >>>> also the more complex. It'll shine in specific cases around large
> > DAGs.
> > >>> If
> > >>>> we have the proper abstractions (methods like DAG.serialize(),
> > >>>> DAG.deser(version)) then I guess that's not an issue.
> > >>>>
> > >>>> Max
> > >>>>
> > >>>> On Fri, Mar 8, 2019 at 5:21 PM Kevin Yang <yr...@gmail.com>
> wrote:
> > >>>>
> > >>>>> Hi Julian, I'm definitely aligned with you guys on making the
> > >> webserver
> > >>>>> independent of DAG parsing, just the end goal to me would be to
> > >> build a
> > >>>>> complete story around serializing DAG--and move with the story in
> > >>> mind. I
> > >>>>> feel like you guys may already have a list of dynamic features we
> > >> need
> > >>> to
> > >>>>> deprecate/change, if that is the case feel free to open the
> > >> discussion
> > >>> on
> > >>>>> what we do to them with DAG serialization.
> > >>>>>
> > >>>>> Julian, Ash, Dan, on 2nd thought I do agree that if we can meet the
> > >>>>> requirements Dan mentioned, it would be nice to have them stored in
> > >> the
> > >>>> DB.
> > >>>>> Some combined solutions like having a column of serialized graph in
> > >> the
> > >>>>> serialized dag table can potentially meet all requirements. What
> > >> format
> > >>>> we
> > >>>>> end up using to represent DAG between components is now less
> > >> important
> > >>>>> IMO--fine to refactor those endpoints only need DagModel to use
> only
> > >>>>> DagModel, easy to do a batch replacement if we decide otherwise
> > >> later.
> > >>>> More
> > >>>>> important is to define this source of truth for serialized DAG.
> > >>>>>
> > >>>>> Ash, ty for the email list, I'll tune my filters accordingly :D I'm
> > >>>> leaning
> > >>>>> towards having a separate process for the parser so we got no
> > >> scheduler
> > >>>>> dependency etc for this parser but we can discuss this in another
> > >>> thread.
> > >>>>>
> > >>>>> On Fri, Mar 8, 2019 at 8:57 AM Dan Davydov
> > >>> <ddavydov@twitter.com.invalid
> > >>>>>
> > >>>>> wrote:
> > >>>>>
> > >>>>>>>
> > >>>>>>> Personally I don’t understand why people are pushing for a
> > >>> JSON-based
> > >>>>> DAG
> > >>>>>>> representation
> > >>>>>>
> > >>>>>> It sounds like you agree that DAGs should be serialized (just in
> > >> the
> > >>> DB
> > >>>>>> instead of JSON), so will only address why JSON is better than
> > >> MySQL
> > >>>> (AKA
> > >>>>>> serializing at the DAG level vs the task level) as far as I can
> > >> see,
> > >>>> and
> > >>>>>> not why we need serialization. If you zoom out and look at all the
> > >>> use
> > >>>>>> cases of serialized DAGs, e.g. having the scheduler use them
> > >> instead
> > >>> of
> > >>>>>> parsing DAGs directly, then it becomes clear that we need all
> > >>>> appropriate
> > >>>>>> metadata in these DAGs, (operator params, DAG properties, etc), in
> > >>>> which
> > >>>>>> case it's not clear how it will fit nicely into a DB table (unless
> > >>> you
> > >>>>>> wanted to do something like (parent_task_id, task_id,
> task_params),
> > >>>> also
> > >>>>>> keep in mind that we will need to store different versions of each
> > >>> DAG
> > >>>> in
> > >>>>>> the future so that we can ensure consistency in a dagrun, i.e.
> each
> > >>>> task
> > >>>>> in
> > >>>>>> a dagrun uses the same version of a DAG.
> > >>>>>>
> > >>>>>> I think some of our requirements should be:
> > >>>>>> 1. The data model will lead to acceptable performance in all of
> its
> > >>>>>> consumers (scheduler, webserver, workers), i.e. no n+1 access
> > >>> patterns
> > >>>>> (my
> > >>>>>> biggest concern about serializing at task level as you propose vs
> > >> at
> > >>>> DAG
> > >>>>>> level)
> > >>>>>> 2. We can have versioning of serialized DAGs
> > >>>>>> 3. The ability to separate DAGs into their own data store (e.g. no
> > >>>>> reliance
> > >>>>>> on joins between the new table and the old one)
> > >>>>>> 4. One source of truth/serialized representation for DAGs
> > >> (currently
> > >>> we
> > >>>>>> have SimpleDAG)
> > >>>>>>
> > >>>>>> If we can full-fill all of these requirements and serialize at the
> > >>> task
> > >>>>>> level rather than the DAG level in the DB, then I agree that
> > >> probably
> > >>>>> makes
> > >>>>>> more sense.
> > >>>>>>
> > >>>>>>
> > >>>>>>> In the proposed PR’s we (Peter, Bas and me) aim to avoid
> > >> re-parsing
> > >>>> DAG
> > >>>>>>> files by querying all the required information from the database.
> > >>> In
> > >>>>> one
> > >>>>>> or
> > >>>>>>> two cases this may however not be possible, in which case we
> > >> might
> > >>>>> either
> > >>>>>>> have to fall back on the DAG file or add the missing information
> > >>> into
> > >>>>> the
> > >>>>>>> database. We can tackle these problems as we encounter them.
> > >>>>>>
> > >>>>>> I think you would have the support of many of committers in
> > >> removing
> > >>>> any
> > >>>>>> use-cases that stand in the way of full serialization, that being
> > >>> said
> > >>>> if
> > >>>>>> we need to remove features we need to do this carefully and
> > >>>> thoughtfully,
> > >>>>>> and ideally with proposed alternatives/work-arounds to cover the
> > >>>>> removals.
> > >>>>>>
> > >>>>>> The counter argument: this PR removes the need for the confusing
> > >>>>> "Refresh"
> > >>>>>>> button from the UI, and in general you only pay the cost for the
> > >>>>>> expensive
> > >>>>>>> DAGs when you ask about them. (I don't know what/when we call the
> > >>>>>>> /pickle_info endpoint of the top of my head)
> > >>>>>>
> > >>>>>> Probably worth splitting out into a separate thread, but I'm
> > >> actually
> > >>>> not
> > >>>>>> sure the refresh button does anything, I think we should double
> > >>>> check...
> > >>>>> I
> > >>>>>> think about 2 years ago there was a commit made that made gunicorn
> > >>>>>> webservers automatically rotate underneath flask (each one would
> > >>>> reparse
> > >>>>>> the DAGbag). Even if it works we should probably remove it since
> > >> the
> > >>>>>> webserver refresh interval is pretty fast, and it just causes
> > >>> confusion
> > >>>>> to
> > >>>>>> users and implies that the DAGs are not refreshed automatically.
> > >>>>>>
> > >>>>>> Do you mean https://json5.org/ or is this a typo? That might be
> > >> okay
> > >>>>> for a
> > >>>>>>> nicer user front end, but the "canonical" version stored in the
> > >> DB
> > >>>>> should
> > >>>>>>> be something "plainer" like just JSON.
> > >>>>>>
> > >>>>>> I think he got this from my reply, and it was just an example, but
> > >>> you
> > >>>>> are
> > >>>>>> right, I agree JSON would be better than JSON5.
> > >>>>>>
> > >>>>>> On Fri, Mar 8, 2019 at 8:53 AM Ash Berlin-Taylor <as...@apache.org>
> > >>>> wrote:
> > >>>>>>
> > >>>>>>> Comments inline.
> > >>>>>>>
> > >>>>>>>> On 8 Mar 2019, at 11:28, Kevin Yang <yr...@gmail.com> wrote:
> > >>>>>>>>
> > >>>>>>>> Hi all,
> > >>>>>>>> When I was preparing some work related to this AIP I found
> > >>>> something
> > >>>>>>> very concerning. I noticed this JIRA ticket <
> > >>>>>>> https://issues.apache.org/jira/browse/AIRFLOW-3562> is trying to
> > >>>>> remove
> > >>>>>>> the dependency of dagbag from webserver, which is awesome--we
> > >>> wanted
> > >>>>>> badly
> > >>>>>>> but never got to start work on. However when I looked at some
> > >>>> subtasks
> > >>>>> of
> > >>>>>>> it, which try to remove dagbag dependency from each endpoint, I
> > >>> found
> > >>>>> the
> > >>>>>>> way we remove the dependency of dagbag is not very ideal. For
> > >>> example
> > >>>>>> this
> > >>>>>>> PR <https://github.com/apache/airflow/pull/4867/files> will
> > >>> require
> > >>>> us
> > >>>>>> to
> > >>>>>>> parse the dag file each time we hit the endpoint.
> > >>>>>>>
> > >>>>>>> The counter argument: this PR removes the need for the confusing
> > >>>>>> "Refresh"
> > >>>>>>> button from the UI, and in general you only pay the cost for the
> > >>>>>> expensive
> > >>>>>>> DAGs when you ask about them. (I don't know what/when we call the
> > >>>>>>> /pickle_info endpoint of the top of my head)
> > >>>>>>>
> > >>>>>>> This end point may be one to hold off on (as it can ask for
> > >>> multiple
> > >>>>>> dags)
> > >>>>>>> but there are some that def don't need a full dag bag or to even
> > >>>> parse
> > >>>>>> the
> > >>>>>>> dag file, the current DAG model has enough info.
> > >>>>>>>
> > >>>>>>>>
> > >>>>>>>>
> > >>>>>>>> If we go down this path, we indeed can get rid of the dagbag
> > >>>>> dependency
> > >>>>>>> easily, but we will have to 1. increase the DB load( not too
> > >>>> concerning
> > >>>>>> at
> > >>>>>>> the moment ), 2. wait the DAG file to be parsed before getting
> > >> the
> > >>>> page
> > >>>>>>> back, potentially multiple times. DAG file can sometimes take
> > >>> quite a
> > >>>>>> while
> > >>>>>>> to parse, e.g. we have some framework DAG files generating large
> > >>>> number
> > >>>>>> of
> > >>>>>>> DAGs from some static config files or even jupyter notebooks and
> > >>> they
> > >>>>> can
> > >>>>>>> take 30+ seconds to parse. Yes we don't like large DAG files but
> > >>>> people
> > >>>>>> do
> > >>>>>>> see the beauty of code as config and sometimes heavily
> > >>> abuseleverage
> > >>>>> it.
> > >>>>>>> Assuming all users have the same nice small python file that can
> > >> be
> > >>>>>> parsed
> > >>>>>>> fast, I'm still a bit worried about this approach. Continuing on
> > >>> this
> > >>>>>> path
> > >>>>>>> means we've chosen DagModel to be the serialized representation
> > >> of
> > >>>> DAG
> > >>>>>> and
> > >>>>>>> DB columns to hold different properties--it can be one candidate
> > >>> but
> > >>>> I
> > >>>>>>> don't know if we should settle on that now. I would personally
> > >>>> prefer a
> > >>>>>>> more compact, e.g. JSON5, and easy to scale representation( such
> > >>> that
> > >>>>>>> serializing new fields != DB upgrade).
> > >>>>>>>
> > >>>>>>> Do you mean https://json5.org/ or is this a typo? That might be
> > >>> okay
> > >>>>> for
> > >>>>>>> a nicer user front end, but the "canonical" version stored in the
> > >>> DB
> > >>>>>> should
> > >>>>>>> be something "plainer" like just JSON.
> > >>>>>>>
> > >>>>>>> I'm not sure that "serializing new fields != DB upgrade" is that
> > >>> big
> > >>>>> of a
> > >>>>>>> concern, as we don't add fields that often. One possible way of
> > >>>> dealing
> > >>>>>>> with it if we do is to have a hybrid approach - a few distinct
> > >>>> columns,
> > >>>>>> but
> > >>>>>>> then a JSON blob. (and if we were only to support postgres we
> > >> could
> > >>>>> just
> > >>>>>>> use JSONb. But I think our friends at Google may object ;) )
> > >>>>>>>
> > >>>>>>> Adding a new column in a DB migration with a default NULL
> > >> shouldn't
> > >>>> be
> > >>>>> an
> > >>>>>>> expensive operation, or difficult to achieve.
> > >>>>>>>
> > >>>>>>>
> > >>>>>>>>
> > >>>>>>>> In my imagination we would have to collect the list of dynamic
> > >>>>> features
> > >>>>>>> depending on unserializable fields of a DAG and start a
> > >>>> discussion/vote
> > >>>>>> on
> > >>>>>>> dropping support of them( I'm working on this but if anyone has
> > >>>> already
> > >>>>>>> done so please take over), decide on the serialized
> > >> representation
> > >>>> of a
> > >>>>>> DAG
> > >>>>>>> and then replace dagbag with it in webserver. Per previous
> > >>> discussion
> > >>>>> and
> > >>>>>>> some offline discussions with Dan, one future of DAG
> > >> serialization
> > >>>>> that I
> > >>>>>>> like would look similar to this:
> > >>>>>>>>
> > >>>>>>>
> > >>>>>>>> https://imgur.com/ncqqQgc
> > >>>>>>>
> > >>>>>>> Something I've thought about before for other things was to embed
> > >>> an
> > >>>>> API
> > >>>>>>> server _into_ the scheduler - this would be useful for k8s
> > >>>>> healthchecks,
> > >>>>>>> native Prometheus metrics without needed statsd bridge, and could
> > >>>> have
> > >>>>>>> endpoints to get information such as this directly.
> > >>>>>>>
> > >>>>>>> I was thinking it would be _in_ the scheduler process using
> > >> either
> > >>>>>> threads
> > >>>>>>> (ick. Python's still got a GIL doesn't it?) or using
> > >> async/twisted
> > >>>> etc.
> > >>>>>>> (not a side-car process like we have with the logs webserver for
> > >>>>> `airflow
> > >>>>>>> worker`).
> > >>>>>>>
> > >>>>>>> (This is possibly an unrelated discussion, but might be worth
> > >>> talking
> > >>>>>>> about?)
> > >>>>>>>
> > >>>>>>>> We can still discuss/vote which approach we want to take but I
> > >>>> don't
> > >>>>>>> want the door to above design to be shut right now or we have to
> > >>>> spend
> > >>>>> a
> > >>>>>>> lot effort switch path later.
> > >>>>>>>>
> > >>>>>>>> Bas and Peter, I'm very sorry to extend the discussion but I do
> > >>>> think
> > >>>>>>> this is tightly related to the AIP and PRs behind it. And my
> > >>> sincere
> > >>>>>>> apology for bringing this up so late( I only pull the open PR
> > >> list
> > >>>>>>> occasionally, if there's a way to subscribe to new PR event I'd
> > >>> love
> > >>>> to
> > >>>>>>> know how).
> > >>>>>>>
> > >>>>>>> It's noisy, but you can subscribe to commits@airflow.apache.org
> > >>> (but
> > >>>>> be
> > >>>>>>> warned, this also includes all Jira tickets, edits of every
> > >> comment
> > >>>> on
> > >>>>>>> github etc.).
> > >>>>>>>
> > >>>>>>>
> > >>>>>>>>
> > >>>>>>>> Cheers,
> > >>>>>>>> Kevin Y
> > >>>>>>>>
> > >>>>>>>> On Thu, Feb 28, 2019 at 1:36 PM Peter van t Hof <
> > >>>>> pjrvanthof@gmail.com
> > >>>>>>> <ma...@gmail.com>> wrote:
> > >>>>>>>> Hi all,
> > >>>>>>>>
> > >>>>>>>> Just some comments one the point Bolke dit give in relation of
> > >> my
> > >>>> PR.
> > >>>>>>>>
> > >>>>>>>> At first, the main focus is: making the webserver stateless.
> > >>>>>>>>
> > >>>>>>>>> 1) Make the webserver stateless: needs the graph of the
> > >>> *current*
> > >>>>> dag
> > >>>>>>>>
> > >>>>>>>> This is the main goal but for this a lot more PR’s will be
> > >> coming
> > >>>>> once
> > >>>>>>> my current is merged. For edges and graph view this is covered in
> > >>> my
> > >>>> PR
> > >>>>>>> already.
> > >>>>>>>>
> > >>>>>>>>> 2) Version dags: for consistency mainly and not requiring
> > >>> parsing
> > >>>>> of
> > >>>>>>> the
> > >>>>>>>>> dag on every loop
> > >>>>>>>>
> > >>>>>>>> In my PR the historical graphs will be stored for each DagRun.
> > >>> This
> > >>>>>>> means that you can see if an older DagRun was the same graph
> > >>>> structure,
> > >>>>>>> even if some tasks does not exists anymore in the current graph.
> > >>>>>> Especially
> > >>>>>>> for dynamic DAG’s this is very useful.
> > >>>>>>>>
> > >>>>>>>>> 3) Make the scheduler not require DAG files. This could be
> > >> done
> > >>>> if
> > >>>>>> the
> > >>>>>>>>> edges contain all information when to trigger the next task.
> > >> We
> > >>>> can
> > >>>>>>> then
> > >>>>>>>>> have event driven dag parsing outside of the scheduler loop,
> > >>> ie.
> > >>>> by
> > >>>>>> the
> > >>>>>>>>> cli. Storage can also be somewhere else (git, artifactory,
> > >>>>>> filesystem,
> > >>>>>>>>> whatever).
> > >>>>>>>>
> > >>>>>>>> The scheduler is almost untouched in this PR. The only thing
> > >> that
> > >>>> is
> > >>>>>>> added is that this edges are saved to the database but the
> > >>> scheduling
> > >>>>>>> itself din’t change. The scheduler depends now still on the DAG
> > >>>> object.
> > >>>>>>>>
> > >>>>>>>>> 4) Fully serialise the dag so it becomes transferable to
> > >>> workers
> > >>>>>>>>
> > >>>>>>>> It nice to see that people has a lot of idea’s about this. But
> > >> as
> > >>>>> Fokko
> > >>>>>>> already mentioned this is out of scope for the issue what we are
> > >>>> trying
> > >>>>>> to
> > >>>>>>> solve. I also have some idea’s about this but I like to limit
> > >> this
> > >>>>> PR/AIP
> > >>>>>>> to the webserver.
> > >>>>>>>>
> > >>>>>>>> For now my PR does solve 1 and 2 and the rest of the behaviour
> > >>>> (like
> > >>>>>>> scheduling) is untouched.
> > >>>>>>>>
> > >>>>>>>> Gr,
> > >>>>>>>> Peter
> > >>>>>>>>
> > >>>>>>>
> > >>>>>>>
> > >>>>>>
> > >>>>>
> > >>>>
> > >>>
> > >>
> >
> >
>
> --
>
> Jarek Potiuk
> Polidea <https://www.polidea.com/> | Principal Software Engineer
>
> M: +48 660 796 129 <+48660796129>
> [image: Polidea] <https://www.polidea.com/>
>

Re: [DISCUSS] AIP-12 Persist DAG into DB

[Jarek Potiuk <Ja...@polidea.com>]

Cool!

On Thu, Jul 18, 2019 at 11:46 AM Ash Berlin-Taylor <as...@apache.org> wrote:

> We didn't reach any conclusion on this yet but I agree, and this is the
> big task that we at Astronomer are going to work on next for Airflow.
>
> I've started chatting to a few of the other committers about this to get a
> an idea of people's priorities, and have had a chat with Alex at Uber about
> their experiences of making their internal fork of Airflow - Piper
> https://eng.uber.com/managing-data-workflows-at-scale/
>
> I'll create something in the wiki (probably not an AIP to start with) to
> collect the possible approaches and downsides/limitations.
>
> Watch this space.
>
> -ash
>
> > On 18 Jul 2019, at 07:05, Tao Feng <fe...@gmail.com> wrote:
> >
> > Do we reach any consensus on this topic /AIP? I think persisting DAG is
> > pretty important actually.
> >
> > -Tao
> >
> > On Tue, Mar 12, 2019 at 3:01 AM Kevin Yang <yr...@gmail.com> wrote:
> >
> >> Hi Fokko,
> >>
> >> As a large cluster maintainer, I’m not a big fan of large DAG files
> >> neither. But I’m not sure if I’ll consider this bad practice. We have
> some
> >> large frameworks, e.g. experimentation and machine learning, that are
> >> complex by nature and generate large number of DAGs from their customer
> >> configs to get better flexibility. I consider them as advance use cases
> of
> >> Airflow and open up a lot potentials for Airflow, unless we’ve
> previously
> >> set some boundaries around how complex DAG codes can be that I’m not
> aware
> >> of. About resulting in an unworkable situation, yes we are experiencing
> >> pain from having such large DAG files, mainly on the webserver side, but
> >> the system overall are running stable. We are actually hoping to improve
> >> the situation by applying solutions like making webserver stateless. It
> is
> >> ok that if the owners of large DAG files need to pay but we should try
> >> minimize the price—longer refresh interval, extra task running time, but
> >> nothing too crazy.
> >>
> >>
> >> I think we’re aligned on storing info in DB as long as we can meet the
> >> requirements Dan mentioned earlier—we just need that balance decided, so
> >> I’m gonna skip this part( out of all the requirements, No.1 seems to be
> >> least clear, maybe we can expand on that). One thing about the proposed
> >> idea is that we implicitly couple DagRun with DAG version, which at the
> >> first glance make sense but imo not very ideal. I feel full versioning
> >> should track all changes instead of tracking changes only when we create
> >> DagRun. E.g. my task failed and I merged new code to fix my task and I
> want
> >> to rerun it with the current code, with serialize DAG during DagRun
> >> creation time we won’t have the up to date snapshot—sure we can work
> around
> >> it by like always keep a current snapshot of DAG but this is kinda messy
> >> and confusing. This is what popped up on the top of my head and w/o full
> >> versioning we might have some other tricky cases, e.g. ur backfill case.
> >> But I just gave a few thoughts into this and you might already have a
> >> complete story that will void my concerns.
> >>
> >>
> >> Cheers,
> >> Kevin Y
> >>
> >> On Sun, Mar 10, 2019 at 11:29 AM Driesprong, Fokko <fokko@driesprong.frl
> >
> >> wrote:
> >>
> >>> Thanks Kevin for opening the discussion. I think it is important to
> have
> >> a
> >>> clear overview on how to approach the AIP.
> >>>
> >>> First of all, how many DAGs do we have that take 30s to parse? I
> consider
> >>> this bad practice, and this would also result in an unworkable
> situation
> >>> with the current setup of Airflow since it will take a lot of resources
> >> on
> >>> the webserver/scheduler, and the whole system will become
> unresponsive. I
> >>> will be hard to cope with such DAGs in general.
> >>>
> >>> The idea from the AIP is to have the versioned version of the dag in
> the
> >>> DB, so in the end, you won't need to parse the whole thing every time.
> >> Only
> >>> when you trigger a DAG, or when you want to see the current status of
> the
> >>> dag.
> >>>
> >>> Like stated earlier, I strongly feel we shouldn't serialize the DAGs as
> >>> JSON(5) or pickles in general. For me, this is deferring the pain of
> >>> setting up a structure of the DAG object itself.
> >>> Having the DAG denormalized in the database will give us cleaner
> storage
> >> of
> >>> our DAG. We can, for example, enforce fields by making them not null,
> so
> >> we
> >>> know that is something is off at write time, instead of read.
> >> Furthermore,
> >>> we're missing logical types such as dates, which we efficiently can
> query
> >>> using the indices of the database.
> >>> Also, with all serialization formats, evolution isn't trivial. Consider
> >> the
> >>> situations when:
> >>> - We're introducing a new field, and it might be null, therefore we
> need
> >> to
> >>> bake in all kinds of logic into the Airflow code, which you don't want.
> >>> With proper migration scripts, you could prefill these fields, and make
> >>> them not null.
> >>> - Changing the models, for example, you still can't change a
> string-type
> >>> into a integer with adding custom logic. In this case, the reviewer
> needs
> >>> to be extra careful that there are no breaking changes introduced.
> Right
> >>> now we're doing minimal forward- and backward compatibilitytesting.
> >>>
> >>> In the case we get too many migrations, we could also squash (some of
> >> them)
> >>> when preparing the release.
> >>>
> >>> Personally, I don't think the serialization is the issue here. As Max
> >>> already mentioned, it is the optimal balance of (de)normalization. From
> >> the
> >>> user perspective, the serialization won't change much of the behaviour
> of
> >>> Airflow.
> >>>
> >>> For me, instead of having `DAG.serialize()` and `DAG.deser(version)` is
> >> not
> >>> the ideal approach. But it might be that we're on the same page :-) I
> >>> believe it should be something like `DagRun.find('fokkos_dag',
> >>> datetime(2018, 03, 01))` and construct the correct version of the dag.
> >>> Since there is an uniqueness constrain on dag_id, datetime, this will
> >>> always return the same dag. You will get the versioned DagRun as it ran
> >>> that time. Serializing the fields adn storing them in the database
> should
> >>> happen transparently when you update the DAG object. When you run a
> dag,
> >>> you'll parse the dag, and then run it. `Dag().create_dagrun(...)`, this
> >>> will create a DagRun as the name suggests, if the version of the dag
> >> still
> >>> exists in the database, it will reuse that one, otherwise it will
> create
> >> a
> >>> new version of the DAG (with all the operators etc). In this sense the
> >>> version of the DAGs should be done within the Dag(Run).
> >>>
> >>> The versioning will change the behavour from a user perspective. Right
> >> now
> >>> we store only a single version. For example, the poor mans backfilling
> >>> won't work anymore. This is clearing the state from past&future, up-
> and
> >>> downstream, and let it catch up again.
> >>> In this case, the old version of the DAG won't exists anymore, and
> >>> potentially there are tasks that aren't in the code anymore. In this
> case
> >>> we need to clear the version of the dag, and rerun it with the latest
> >>> version `DagRun.find('fokkos_dag', datetime(2018, 03, 01)).clear()`.
> How
> >> we
> >>> are going to do clear's downstram in the middle of the dag, that is
> >>> something I still have to figure out. Because potentially there are
> tasks
> >>> that can't be rerun because the underlying Python code has changed,
> both
> >> on
> >>> user level as on Airflow level. It will be impossible to get these
> >> features
> >>> pure in that sense.
> >>> I would not suggest adding a new status in here, indicating that the
> task
> >>> can't be rerun since it isn't part of the DAG anymore. We have to find
> >> the
> >>> balance here in adding complexity (also to the scheduler) and features
> >> that
> >>> we need to introduce to help the user.
> >>>
> >>> Cheers, Fokko
> >>>
> >>> Ps. Jarek, interesting idea. It shouldn't be too hard to make Airflow
> >> more
> >>> k8s native. You could package your dags within your container, and do a
> >>> rolling update. Add the DAGs as the last layer, and then point the DAGs
> >>> folder to the same location. The hard part here is that you need to
> >>> gracefuly restart the workers. Currently AFAIK the signals given to the
> >> pod
> >>> aren't respected. So when the scheduler/webserver/worker receives a
> >>> SIGTERM, it should stop the jobs nicely and then exit the container,
> >> before
> >>> k8s kills the container using a SIGKILL.  This will be challenging with
> >> the
> >>> workers, which they are potentially long-running. Maybe stop kicking
> off
> >>> new jobs, and let the old ones finish, will be good enough, but then we
> >>> need to increase the standard kill timeout substantially. Having this
> >> would
> >>> also enable the autoscaling of the workers.
> >>>
> >>>
> >>>
> >>> Op za 9 mrt. 2019 om 19:07 schreef Maxime Beauchemin <
> >>> maximebeauchemin@gmail.com>:
> >>>
> >>>> I want to raise the question of the amount of normalization we want to
> >>> use
> >>>> here as it seems the to be an area that needs more attention.
> >>>>
> >>>> The SIP suggest having DAG blobs, task blobs and edges (call it the
> >>>> fairly-normalized). I also like the idea of all-encompassing (call it
> >>>> very-denormalized) DAG blobs as it seems easier to manage in terms of
> >>>> versioning. The question here is whether we go with one of these
> method
> >>>> exclusively, something in-between or even a hybrid approach (redundant
> >>>> blobs that use different level of normalization).
> >>>>
> >>>> It's nice and simple to just push or pull DAG atomic objects with a
> >>> version
> >>>> stamp on it. It's clearly simpler than dealing with 3 versioned tables
> >>>> (dag, tasks, edges). There are a lot of pros/cons, and they become
> more
> >>>> apparent with the perspective of very large DAGs. If the web server is
> >>>> building a "task details page", using the "fairly-normalized" model,
> it
> >>> can
> >>>> just pull what it needs instead of pulling the large DAG blob.
> >> Similarly,
> >>>> if building a sub-tree view (a subset of the DAG), perhaps it can only
> >>>> retrieve what it needs. But if you need the whole DAG (say for the
> >>>> scheduler use case) then you're dealing with more complex SQL/ORM
> >>>> operations (joins hopefully, or multiple db round trips)
> >>>>
> >>>> Now maybe the right approach is more something like 2 tables: DAG and
> >>>> task_details, where edges keys are denormalized into DAG (arguably
> >>> that's a
> >>>> few KBs at most, even for large DAGs), and maybe the DAG object has
> >> most
> >>> of
> >>>> the high level task metadata information (operator, name, baseoperator
> >>> key
> >>>> attrs), and task_details has the big blobs (SQL code). This is
> >> probably a
> >>>> nice compromise, the question becomes "how much task-level detail do
> we
> >>>> store in the DAG-centric blog?", probably not much to keep the DAG
> >>> objects
> >>>> as small as possible. The main downside here is that you cannot have
> >> the
> >>>> database join and have to do 2 round trips to reconstruct a DAG object
> >>>> (fetch the DAG, parse the object to get the list of tasks, and then
> run
> >>>> another db query to get those task details).
> >>>>
> >>>> To resume, I'd qualify the more normalized approach as the most
> proper,
> >>> but
> >>>> also the more complex. It'll shine in specific cases around large
> DAGs.
> >>> If
> >>>> we have the proper abstractions (methods like DAG.serialize(),
> >>>> DAG.deser(version)) then I guess that's not an issue.
> >>>>
> >>>> Max
> >>>>
> >>>> On Fri, Mar 8, 2019 at 5:21 PM Kevin Yang <yr...@gmail.com> wrote:
> >>>>
> >>>>> Hi Julian, I'm definitely aligned with you guys on making the
> >> webserver
> >>>>> independent of DAG parsing, just the end goal to me would be to
> >> build a
> >>>>> complete story around serializing DAG--and move with the story in
> >>> mind. I
> >>>>> feel like you guys may already have a list of dynamic features we
> >> need
> >>> to
> >>>>> deprecate/change, if that is the case feel free to open the
> >> discussion
> >>> on
> >>>>> what we do to them with DAG serialization.
> >>>>>
> >>>>> Julian, Ash, Dan, on 2nd thought I do agree that if we can meet the
> >>>>> requirements Dan mentioned, it would be nice to have them stored in
> >> the
> >>>> DB.
> >>>>> Some combined solutions like having a column of serialized graph in
> >> the
> >>>>> serialized dag table can potentially meet all requirements. What
> >> format
> >>>> we
> >>>>> end up using to represent DAG between components is now less
> >> important
> >>>>> IMO--fine to refactor those endpoints only need DagModel to use only
> >>>>> DagModel, easy to do a batch replacement if we decide otherwise
> >> later.
> >>>> More
> >>>>> important is to define this source of truth for serialized DAG.
> >>>>>
> >>>>> Ash, ty for the email list, I'll tune my filters accordingly :D I'm
> >>>> leaning
> >>>>> towards having a separate process for the parser so we got no
> >> scheduler
> >>>>> dependency etc for this parser but we can discuss this in another
> >>> thread.
> >>>>>
> >>>>> On Fri, Mar 8, 2019 at 8:57 AM Dan Davydov
> >>> <ddavydov@twitter.com.invalid
> >>>>>
> >>>>> wrote:
> >>>>>
> >>>>>>>
> >>>>>>> Personally I don’t understand why people are pushing for a
> >>> JSON-based
> >>>>> DAG
> >>>>>>> representation
> >>>>>>
> >>>>>> It sounds like you agree that DAGs should be serialized (just in
> >> the
> >>> DB
> >>>>>> instead of JSON), so will only address why JSON is better than
> >> MySQL
> >>>> (AKA
> >>>>>> serializing at the DAG level vs the task level) as far as I can
> >> see,
> >>>> and
> >>>>>> not why we need serialization. If you zoom out and look at all the
> >>> use
> >>>>>> cases of serialized DAGs, e.g. having the scheduler use them
> >> instead
> >>> of
> >>>>>> parsing DAGs directly, then it becomes clear that we need all
> >>>> appropriate
> >>>>>> metadata in these DAGs, (operator params, DAG properties, etc), in
> >>>> which
> >>>>>> case it's not clear how it will fit nicely into a DB table (unless
> >>> you
> >>>>>> wanted to do something like (parent_task_id, task_id, task_params),
> >>>> also
> >>>>>> keep in mind that we will need to store different versions of each
> >>> DAG
> >>>> in
> >>>>>> the future so that we can ensure consistency in a dagrun, i.e. each
> >>>> task
> >>>>> in
> >>>>>> a dagrun uses the same version of a DAG.
> >>>>>>
> >>>>>> I think some of our requirements should be:
> >>>>>> 1. The data model will lead to acceptable performance in all of its
> >>>>>> consumers (scheduler, webserver, workers), i.e. no n+1 access
> >>> patterns
> >>>>> (my
> >>>>>> biggest concern about serializing at task level as you propose vs
> >> at
> >>>> DAG
> >>>>>> level)
> >>>>>> 2. We can have versioning of serialized DAGs
> >>>>>> 3. The ability to separate DAGs into their own data store (e.g. no
> >>>>> reliance
> >>>>>> on joins between the new table and the old one)
> >>>>>> 4. One source of truth/serialized representation for DAGs
> >> (currently
> >>> we
> >>>>>> have SimpleDAG)
> >>>>>>
> >>>>>> If we can full-fill all of these requirements and serialize at the
> >>> task
> >>>>>> level rather than the DAG level in the DB, then I agree that
> >> probably
> >>>>> makes
> >>>>>> more sense.
> >>>>>>
> >>>>>>
> >>>>>>> In the proposed PR’s we (Peter, Bas and me) aim to avoid
> >> re-parsing
> >>>> DAG
> >>>>>>> files by querying all the required information from the database.
> >>> In
> >>>>> one
> >>>>>> or
> >>>>>>> two cases this may however not be possible, in which case we
> >> might
> >>>>> either
> >>>>>>> have to fall back on the DAG file or add the missing information
> >>> into
> >>>>> the
> >>>>>>> database. We can tackle these problems as we encounter them.
> >>>>>>
> >>>>>> I think you would have the support of many of committers in
> >> removing
> >>>> any
> >>>>>> use-cases that stand in the way of full serialization, that being
> >>> said
> >>>> if
> >>>>>> we need to remove features we need to do this carefully and
> >>>> thoughtfully,
> >>>>>> and ideally with proposed alternatives/work-arounds to cover the
> >>>>> removals.
> >>>>>>
> >>>>>> The counter argument: this PR removes the need for the confusing
> >>>>> "Refresh"
> >>>>>>> button from the UI, and in general you only pay the cost for the
> >>>>>> expensive
> >>>>>>> DAGs when you ask about them. (I don't know what/when we call the
> >>>>>>> /pickle_info endpoint of the top of my head)
> >>>>>>
> >>>>>> Probably worth splitting out into a separate thread, but I'm
> >> actually
> >>>> not
> >>>>>> sure the refresh button does anything, I think we should double
> >>>> check...
> >>>>> I
> >>>>>> think about 2 years ago there was a commit made that made gunicorn
> >>>>>> webservers automatically rotate underneath flask (each one would
> >>>> reparse
> >>>>>> the DAGbag). Even if it works we should probably remove it since
> >> the
> >>>>>> webserver refresh interval is pretty fast, and it just causes
> >>> confusion
> >>>>> to
> >>>>>> users and implies that the DAGs are not refreshed automatically.
> >>>>>>
> >>>>>> Do you mean https://json5.org/ or is this a typo? That might be
> >> okay
> >>>>> for a
> >>>>>>> nicer user front end, but the "canonical" version stored in the
> >> DB
> >>>>> should
> >>>>>>> be something "plainer" like just JSON.
> >>>>>>
> >>>>>> I think he got this from my reply, and it was just an example, but
> >>> you
> >>>>> are
> >>>>>> right, I agree JSON would be better than JSON5.
> >>>>>>
> >>>>>> On Fri, Mar 8, 2019 at 8:53 AM Ash Berlin-Taylor <as...@apache.org>
> >>>> wrote:
> >>>>>>
> >>>>>>> Comments inline.
> >>>>>>>
> >>>>>>>> On 8 Mar 2019, at 11:28, Kevin Yang <yr...@gmail.com> wrote:
> >>>>>>>>
> >>>>>>>> Hi all,
> >>>>>>>> When I was preparing some work related to this AIP I found
> >>>> something
> >>>>>>> very concerning. I noticed this JIRA ticket <
> >>>>>>> https://issues.apache.org/jira/browse/AIRFLOW-3562> is trying to
> >>>>> remove
> >>>>>>> the dependency of dagbag from webserver, which is awesome--we
> >>> wanted
> >>>>>> badly
> >>>>>>> but never got to start work on. However when I looked at some
> >>>> subtasks
> >>>>> of
> >>>>>>> it, which try to remove dagbag dependency from each endpoint, I
> >>> found
> >>>>> the
> >>>>>>> way we remove the dependency of dagbag is not very ideal. For
> >>> example
> >>>>>> this
> >>>>>>> PR <https://github.com/apache/airflow/pull/4867/files> will
> >>> require
> >>>> us
> >>>>>> to
> >>>>>>> parse the dag file each time we hit the endpoint.
> >>>>>>>
> >>>>>>> The counter argument: this PR removes the need for the confusing
> >>>>>> "Refresh"
> >>>>>>> button from the UI, and in general you only pay the cost for the
> >>>>>> expensive
> >>>>>>> DAGs when you ask about them. (I don't know what/when we call the
> >>>>>>> /pickle_info endpoint of the top of my head)
> >>>>>>>
> >>>>>>> This end point may be one to hold off on (as it can ask for
> >>> multiple
> >>>>>> dags)
> >>>>>>> but there are some that def don't need a full dag bag or to even
> >>>> parse
> >>>>>> the
> >>>>>>> dag file, the current DAG model has enough info.
> >>>>>>>
> >>>>>>>>
> >>>>>>>>
> >>>>>>>> If we go down this path, we indeed can get rid of the dagbag
> >>>>> dependency
> >>>>>>> easily, but we will have to 1. increase the DB load( not too
> >>>> concerning
> >>>>>> at
> >>>>>>> the moment ), 2. wait the DAG file to be parsed before getting
> >> the
> >>>> page
> >>>>>>> back, potentially multiple times. DAG file can sometimes take
> >>> quite a
> >>>>>> while
> >>>>>>> to parse, e.g. we have some framework DAG files generating large
> >>>> number
> >>>>>> of
> >>>>>>> DAGs from some static config files or even jupyter notebooks and
> >>> they
> >>>>> can
> >>>>>>> take 30+ seconds to parse. Yes we don't like large DAG files but
> >>>> people
> >>>>>> do
> >>>>>>> see the beauty of code as config and sometimes heavily
> >>> abuseleverage
> >>>>> it.
> >>>>>>> Assuming all users have the same nice small python file that can
> >> be
> >>>>>> parsed
> >>>>>>> fast, I'm still a bit worried about this approach. Continuing on
> >>> this
> >>>>>> path
> >>>>>>> means we've chosen DagModel to be the serialized representation
> >> of
> >>>> DAG
> >>>>>> and
> >>>>>>> DB columns to hold different properties--it can be one candidate
> >>> but
> >>>> I
> >>>>>>> don't know if we should settle on that now. I would personally
> >>>> prefer a
> >>>>>>> more compact, e.g. JSON5, and easy to scale representation( such
> >>> that
> >>>>>>> serializing new fields != DB upgrade).
> >>>>>>>
> >>>>>>> Do you mean https://json5.org/ or is this a typo? That might be
> >>> okay
> >>>>> for
> >>>>>>> a nicer user front end, but the "canonical" version stored in the
> >>> DB
> >>>>>> should
> >>>>>>> be something "plainer" like just JSON.
> >>>>>>>
> >>>>>>> I'm not sure that "serializing new fields != DB upgrade" is that
> >>> big
> >>>>> of a
> >>>>>>> concern, as we don't add fields that often. One possible way of
> >>>> dealing
> >>>>>>> with it if we do is to have a hybrid approach - a few distinct
> >>>> columns,
> >>>>>> but
> >>>>>>> then a JSON blob. (and if we were only to support postgres we
> >> could
> >>>>> just
> >>>>>>> use JSONb. But I think our friends at Google may object ;) )
> >>>>>>>
> >>>>>>> Adding a new column in a DB migration with a default NULL
> >> shouldn't
> >>>> be
> >>>>> an
> >>>>>>> expensive operation, or difficult to achieve.
> >>>>>>>
> >>>>>>>
> >>>>>>>>
> >>>>>>>> In my imagination we would have to collect the list of dynamic
> >>>>> features
> >>>>>>> depending on unserializable fields of a DAG and start a
> >>>> discussion/vote
> >>>>>> on
> >>>>>>> dropping support of them( I'm working on this but if anyone has
> >>>> already
> >>>>>>> done so please take over), decide on the serialized
> >> representation
> >>>> of a
> >>>>>> DAG
> >>>>>>> and then replace dagbag with it in webserver. Per previous
> >>> discussion
> >>>>> and
> >>>>>>> some offline discussions with Dan, one future of DAG
> >> serialization
> >>>>> that I
> >>>>>>> like would look similar to this:
> >>>>>>>>
> >>>>>>>
> >>>>>>>> https://imgur.com/ncqqQgc
> >>>>>>>
> >>>>>>> Something I've thought about before for other things was to embed
> >>> an
> >>>>> API
> >>>>>>> server _into_ the scheduler - this would be useful for k8s
> >>>>> healthchecks,
> >>>>>>> native Prometheus metrics without needed statsd bridge, and could
> >>>> have
> >>>>>>> endpoints to get information such as this directly.
> >>>>>>>
> >>>>>>> I was thinking it would be _in_ the scheduler process using
> >> either
> >>>>>> threads
> >>>>>>> (ick. Python's still got a GIL doesn't it?) or using
> >> async/twisted
> >>>> etc.
> >>>>>>> (not a side-car process like we have with the logs webserver for
> >>>>> `airflow
> >>>>>>> worker`).
> >>>>>>>
> >>>>>>> (This is possibly an unrelated discussion, but might be worth
> >>> talking
> >>>>>>> about?)
> >>>>>>>
> >>>>>>>> We can still discuss/vote which approach we want to take but I
> >>>> don't
> >>>>>>> want the door to above design to be shut right now or we have to
> >>>> spend
> >>>>> a
> >>>>>>> lot effort switch path later.
> >>>>>>>>
> >>>>>>>> Bas and Peter, I'm very sorry to extend the discussion but I do
> >>>> think
> >>>>>>> this is tightly related to the AIP and PRs behind it. And my
> >>> sincere
> >>>>>>> apology for bringing this up so late( I only pull the open PR
> >> list
> >>>>>>> occasionally, if there's a way to subscribe to new PR event I'd
> >>> love
> >>>> to
> >>>>>>> know how).
> >>>>>>>
> >>>>>>> It's noisy, but you can subscribe to commits@airflow.apache.org
> >>> (but
> >>>>> be
> >>>>>>> warned, this also includes all Jira tickets, edits of every
> >> comment
> >>>> on
> >>>>>>> github etc.).
> >>>>>>>
> >>>>>>>
> >>>>>>>>
> >>>>>>>> Cheers,
> >>>>>>>> Kevin Y
> >>>>>>>>
> >>>>>>>> On Thu, Feb 28, 2019 at 1:36 PM Peter van t Hof <
> >>>>> pjrvanthof@gmail.com
> >>>>>>> <ma...@gmail.com>> wrote:
> >>>>>>>> Hi all,
> >>>>>>>>
> >>>>>>>> Just some comments one the point Bolke dit give in relation of
> >> my
> >>>> PR.
> >>>>>>>>
> >>>>>>>> At first, the main focus is: making the webserver stateless.
> >>>>>>>>
> >>>>>>>>> 1) Make the webserver stateless: needs the graph of the
> >>> *current*
> >>>>> dag
> >>>>>>>>
> >>>>>>>> This is the main goal but for this a lot more PR’s will be
> >> coming
> >>>>> once
> >>>>>>> my current is merged. For edges and graph view this is covered in
> >>> my
> >>>> PR
> >>>>>>> already.
> >>>>>>>>
> >>>>>>>>> 2) Version dags: for consistency mainly and not requiring
> >>> parsing
> >>>>> of
> >>>>>>> the
> >>>>>>>>> dag on every loop
> >>>>>>>>
> >>>>>>>> In my PR the historical graphs will be stored for each DagRun.
> >>> This
> >>>>>>> means that you can see if an older DagRun was the same graph
> >>>> structure,
> >>>>>>> even if some tasks does not exists anymore in the current graph.
> >>>>>> Especially
> >>>>>>> for dynamic DAG’s this is very useful.
> >>>>>>>>
> >>>>>>>>> 3) Make the scheduler not require DAG files. This could be
> >> done
> >>>> if
> >>>>>> the
> >>>>>>>>> edges contain all information when to trigger the next task.
> >> We
> >>>> can
> >>>>>>> then
> >>>>>>>>> have event driven dag parsing outside of the scheduler loop,
> >>> ie.
> >>>> by
> >>>>>> the
> >>>>>>>>> cli. Storage can also be somewhere else (git, artifactory,
> >>>>>> filesystem,
> >>>>>>>>> whatever).
> >>>>>>>>
> >>>>>>>> The scheduler is almost untouched in this PR. The only thing
> >> that
> >>>> is
> >>>>>>> added is that this edges are saved to the database but the
> >>> scheduling
> >>>>>>> itself din’t change. The scheduler depends now still on the DAG
> >>>> object.
> >>>>>>>>
> >>>>>>>>> 4) Fully serialise the dag so it becomes transferable to
> >>> workers
> >>>>>>>>
> >>>>>>>> It nice to see that people has a lot of idea’s about this. But
> >> as
> >>>>> Fokko
> >>>>>>> already mentioned this is out of scope for the issue what we are
> >>>> trying
> >>>>>> to
> >>>>>>> solve. I also have some idea’s about this but I like to limit
> >> this
> >>>>> PR/AIP
> >>>>>>> to the webserver.
> >>>>>>>>
> >>>>>>>> For now my PR does solve 1 and 2 and the rest of the behaviour
> >>>> (like
> >>>>>>> scheduling) is untouched.
> >>>>>>>>
> >>>>>>>> Gr,
> >>>>>>>> Peter
> >>>>>>>>
> >>>>>>>
> >>>>>>>
> >>>>>>
> >>>>>
> >>>>
> >>>
> >>
>
>

-- 

Jarek Potiuk
Polidea <https://www.polidea.com/> | Principal Software Engineer

M: +48 660 796 129 <+48660796129>
[image: Polidea] <https://www.polidea.com/>

Re: [DISCUSS] AIP-12 Persist DAG into DB

[Ash Berlin-Taylor <as...@apache.org>]

We didn't reach any conclusion on this yet but I agree, and this is the big task that we at Astronomer are going to work on next for Airflow.

I've started chatting to a few of the other committers about this to get a an idea of people's priorities, and have had a chat with Alex at Uber about their experiences of making their internal fork of Airflow - Piper https://eng.uber.com/managing-data-workflows-at-scale/

I'll create something in the wiki (probably not an AIP to start with) to collect the possible approaches and downsides/limitations.

Watch this space.

-ash

> On 18 Jul 2019, at 07:05, Tao Feng <fe...@gmail.com> wrote:
> 
> Do we reach any consensus on this topic /AIP? I think persisting DAG is
> pretty important actually.
> 
> -Tao
> 
> On Tue, Mar 12, 2019 at 3:01 AM Kevin Yang <yr...@gmail.com> wrote:
> 
>> Hi Fokko,
>> 
>> As a large cluster maintainer, I’m not a big fan of large DAG files
>> neither. But I’m not sure if I’ll consider this bad practice. We have some
>> large frameworks, e.g. experimentation and machine learning, that are
>> complex by nature and generate large number of DAGs from their customer
>> configs to get better flexibility. I consider them as advance use cases of
>> Airflow and open up a lot potentials for Airflow, unless we’ve previously
>> set some boundaries around how complex DAG codes can be that I’m not aware
>> of. About resulting in an unworkable situation, yes we are experiencing
>> pain from having such large DAG files, mainly on the webserver side, but
>> the system overall are running stable. We are actually hoping to improve
>> the situation by applying solutions like making webserver stateless. It is
>> ok that if the owners of large DAG files need to pay but we should try
>> minimize the price—longer refresh interval, extra task running time, but
>> nothing too crazy.
>> 
>> 
>> I think we’re aligned on storing info in DB as long as we can meet the
>> requirements Dan mentioned earlier—we just need that balance decided, so
>> I’m gonna skip this part( out of all the requirements, No.1 seems to be
>> least clear, maybe we can expand on that). One thing about the proposed
>> idea is that we implicitly couple DagRun with DAG version, which at the
>> first glance make sense but imo not very ideal. I feel full versioning
>> should track all changes instead of tracking changes only when we create
>> DagRun. E.g. my task failed and I merged new code to fix my task and I want
>> to rerun it with the current code, with serialize DAG during DagRun
>> creation time we won’t have the up to date snapshot—sure we can work around
>> it by like always keep a current snapshot of DAG but this is kinda messy
>> and confusing. This is what popped up on the top of my head and w/o full
>> versioning we might have some other tricky cases, e.g. ur backfill case.
>> But I just gave a few thoughts into this and you might already have a
>> complete story that will void my concerns.
>> 
>> 
>> Cheers,
>> Kevin Y
>> 
>> On Sun, Mar 10, 2019 at 11:29 AM Driesprong, Fokko <fo...@driesprong.frl>
>> wrote:
>> 
>>> Thanks Kevin for opening the discussion. I think it is important to have
>> a
>>> clear overview on how to approach the AIP.
>>> 
>>> First of all, how many DAGs do we have that take 30s to parse? I consider
>>> this bad practice, and this would also result in an unworkable situation
>>> with the current setup of Airflow since it will take a lot of resources
>> on
>>> the webserver/scheduler, and the whole system will become unresponsive. I
>>> will be hard to cope with such DAGs in general.
>>> 
>>> The idea from the AIP is to have the versioned version of the dag in the
>>> DB, so in the end, you won't need to parse the whole thing every time.
>> Only
>>> when you trigger a DAG, or when you want to see the current status of the
>>> dag.
>>> 
>>> Like stated earlier, I strongly feel we shouldn't serialize the DAGs as
>>> JSON(5) or pickles in general. For me, this is deferring the pain of
>>> setting up a structure of the DAG object itself.
>>> Having the DAG denormalized in the database will give us cleaner storage
>> of
>>> our DAG. We can, for example, enforce fields by making them not null, so
>> we
>>> know that is something is off at write time, instead of read.
>> Furthermore,
>>> we're missing logical types such as dates, which we efficiently can query
>>> using the indices of the database.
>>> Also, with all serialization formats, evolution isn't trivial. Consider
>> the
>>> situations when:
>>> - We're introducing a new field, and it might be null, therefore we need
>> to
>>> bake in all kinds of logic into the Airflow code, which you don't want.
>>> With proper migration scripts, you could prefill these fields, and make
>>> them not null.
>>> - Changing the models, for example, you still can't change a string-type
>>> into a integer with adding custom logic. In this case, the reviewer needs
>>> to be extra careful that there are no breaking changes introduced. Right
>>> now we're doing minimal forward- and backward compatibilitytesting.
>>> 
>>> In the case we get too many migrations, we could also squash (some of
>> them)
>>> when preparing the release.
>>> 
>>> Personally, I don't think the serialization is the issue here. As Max
>>> already mentioned, it is the optimal balance of (de)normalization. From
>> the
>>> user perspective, the serialization won't change much of the behaviour of
>>> Airflow.
>>> 
>>> For me, instead of having `DAG.serialize()` and `DAG.deser(version)` is
>> not
>>> the ideal approach. But it might be that we're on the same page :-) I
>>> believe it should be something like `DagRun.find('fokkos_dag',
>>> datetime(2018, 03, 01))` and construct the correct version of the dag.
>>> Since there is an uniqueness constrain on dag_id, datetime, this will
>>> always return the same dag. You will get the versioned DagRun as it ran
>>> that time. Serializing the fields adn storing them in the database should
>>> happen transparently when you update the DAG object. When you run a dag,
>>> you'll parse the dag, and then run it. `Dag().create_dagrun(...)`, this
>>> will create a DagRun as the name suggests, if the version of the dag
>> still
>>> exists in the database, it will reuse that one, otherwise it will create
>> a
>>> new version of the DAG (with all the operators etc). In this sense the
>>> version of the DAGs should be done within the Dag(Run).
>>> 
>>> The versioning will change the behavour from a user perspective. Right
>> now
>>> we store only a single version. For example, the poor mans backfilling
>>> won't work anymore. This is clearing the state from past&future, up- and
>>> downstream, and let it catch up again.
>>> In this case, the old version of the DAG won't exists anymore, and
>>> potentially there are tasks that aren't in the code anymore. In this case
>>> we need to clear the version of the dag, and rerun it with the latest
>>> version `DagRun.find('fokkos_dag', datetime(2018, 03, 01)).clear()`. How
>> we
>>> are going to do clear's downstram in the middle of the dag, that is
>>> something I still have to figure out. Because potentially there are tasks
>>> that can't be rerun because the underlying Python code has changed, both
>> on
>>> user level as on Airflow level. It will be impossible to get these
>> features
>>> pure in that sense.
>>> I would not suggest adding a new status in here, indicating that the task
>>> can't be rerun since it isn't part of the DAG anymore. We have to find
>> the
>>> balance here in adding complexity (also to the scheduler) and features
>> that
>>> we need to introduce to help the user.
>>> 
>>> Cheers, Fokko
>>> 
>>> Ps. Jarek, interesting idea. It shouldn't be too hard to make Airflow
>> more
>>> k8s native. You could package your dags within your container, and do a
>>> rolling update. Add the DAGs as the last layer, and then point the DAGs
>>> folder to the same location. The hard part here is that you need to
>>> gracefuly restart the workers. Currently AFAIK the signals given to the
>> pod
>>> aren't respected. So when the scheduler/webserver/worker receives a
>>> SIGTERM, it should stop the jobs nicely and then exit the container,
>> before
>>> k8s kills the container using a SIGKILL.  This will be challenging with
>> the
>>> workers, which they are potentially long-running. Maybe stop kicking off
>>> new jobs, and let the old ones finish, will be good enough, but then we
>>> need to increase the standard kill timeout substantially. Having this
>> would
>>> also enable the autoscaling of the workers.
>>> 
>>> 
>>> 
>>> Op za 9 mrt. 2019 om 19:07 schreef Maxime Beauchemin <
>>> maximebeauchemin@gmail.com>:
>>> 
>>>> I want to raise the question of the amount of normalization we want to
>>> use
>>>> here as it seems the to be an area that needs more attention.
>>>> 
>>>> The SIP suggest having DAG blobs, task blobs and edges (call it the
>>>> fairly-normalized). I also like the idea of all-encompassing (call it
>>>> very-denormalized) DAG blobs as it seems easier to manage in terms of
>>>> versioning. The question here is whether we go with one of these method
>>>> exclusively, something in-between or even a hybrid approach (redundant
>>>> blobs that use different level of normalization).
>>>> 
>>>> It's nice and simple to just push or pull DAG atomic objects with a
>>> version
>>>> stamp on it. It's clearly simpler than dealing with 3 versioned tables
>>>> (dag, tasks, edges). There are a lot of pros/cons, and they become more
>>>> apparent with the perspective of very large DAGs. If the web server is
>>>> building a "task details page", using the "fairly-normalized" model, it
>>> can
>>>> just pull what it needs instead of pulling the large DAG blob.
>> Similarly,
>>>> if building a sub-tree view (a subset of the DAG), perhaps it can only
>>>> retrieve what it needs. But if you need the whole DAG (say for the
>>>> scheduler use case) then you're dealing with more complex SQL/ORM
>>>> operations (joins hopefully, or multiple db round trips)
>>>> 
>>>> Now maybe the right approach is more something like 2 tables: DAG and
>>>> task_details, where edges keys are denormalized into DAG (arguably
>>> that's a
>>>> few KBs at most, even for large DAGs), and maybe the DAG object has
>> most
>>> of
>>>> the high level task metadata information (operator, name, baseoperator
>>> key
>>>> attrs), and task_details has the big blobs (SQL code). This is
>> probably a
>>>> nice compromise, the question becomes "how much task-level detail do we
>>>> store in the DAG-centric blog?", probably not much to keep the DAG
>>> objects
>>>> as small as possible. The main downside here is that you cannot have
>> the
>>>> database join and have to do 2 round trips to reconstruct a DAG object
>>>> (fetch the DAG, parse the object to get the list of tasks, and then run
>>>> another db query to get those task details).
>>>> 
>>>> To resume, I'd qualify the more normalized approach as the most proper,
>>> but
>>>> also the more complex. It'll shine in specific cases around large DAGs.
>>> If
>>>> we have the proper abstractions (methods like DAG.serialize(),
>>>> DAG.deser(version)) then I guess that's not an issue.
>>>> 
>>>> Max
>>>> 
>>>> On Fri, Mar 8, 2019 at 5:21 PM Kevin Yang <yr...@gmail.com> wrote:
>>>> 
>>>>> Hi Julian, I'm definitely aligned with you guys on making the
>> webserver
>>>>> independent of DAG parsing, just the end goal to me would be to
>> build a
>>>>> complete story around serializing DAG--and move with the story in
>>> mind. I
>>>>> feel like you guys may already have a list of dynamic features we
>> need
>>> to
>>>>> deprecate/change, if that is the case feel free to open the
>> discussion
>>> on
>>>>> what we do to them with DAG serialization.
>>>>> 
>>>>> Julian, Ash, Dan, on 2nd thought I do agree that if we can meet the
>>>>> requirements Dan mentioned, it would be nice to have them stored in
>> the
>>>> DB.
>>>>> Some combined solutions like having a column of serialized graph in
>> the
>>>>> serialized dag table can potentially meet all requirements. What
>> format
>>>> we
>>>>> end up using to represent DAG between components is now less
>> important
>>>>> IMO--fine to refactor those endpoints only need DagModel to use only
>>>>> DagModel, easy to do a batch replacement if we decide otherwise
>> later.
>>>> More
>>>>> important is to define this source of truth for serialized DAG.
>>>>> 
>>>>> Ash, ty for the email list, I'll tune my filters accordingly :D I'm
>>>> leaning
>>>>> towards having a separate process for the parser so we got no
>> scheduler
>>>>> dependency etc for this parser but we can discuss this in another
>>> thread.
>>>>> 
>>>>> On Fri, Mar 8, 2019 at 8:57 AM Dan Davydov
>>> <ddavydov@twitter.com.invalid
>>>>> 
>>>>> wrote:
>>>>> 
>>>>>>> 
>>>>>>> Personally I don’t understand why people are pushing for a
>>> JSON-based
>>>>> DAG
>>>>>>> representation
>>>>>> 
>>>>>> It sounds like you agree that DAGs should be serialized (just in
>> the
>>> DB
>>>>>> instead of JSON), so will only address why JSON is better than
>> MySQL
>>>> (AKA
>>>>>> serializing at the DAG level vs the task level) as far as I can
>> see,
>>>> and
>>>>>> not why we need serialization. If you zoom out and look at all the
>>> use
>>>>>> cases of serialized DAGs, e.g. having the scheduler use them
>> instead
>>> of
>>>>>> parsing DAGs directly, then it becomes clear that we need all
>>>> appropriate
>>>>>> metadata in these DAGs, (operator params, DAG properties, etc), in
>>>> which
>>>>>> case it's not clear how it will fit nicely into a DB table (unless
>>> you
>>>>>> wanted to do something like (parent_task_id, task_id, task_params),
>>>> also
>>>>>> keep in mind that we will need to store different versions of each
>>> DAG
>>>> in
>>>>>> the future so that we can ensure consistency in a dagrun, i.e. each
>>>> task
>>>>> in
>>>>>> a dagrun uses the same version of a DAG.
>>>>>> 
>>>>>> I think some of our requirements should be:
>>>>>> 1. The data model will lead to acceptable performance in all of its
>>>>>> consumers (scheduler, webserver, workers), i.e. no n+1 access
>>> patterns
>>>>> (my
>>>>>> biggest concern about serializing at task level as you propose vs
>> at
>>>> DAG
>>>>>> level)
>>>>>> 2. We can have versioning of serialized DAGs
>>>>>> 3. The ability to separate DAGs into their own data store (e.g. no
>>>>> reliance
>>>>>> on joins between the new table and the old one)
>>>>>> 4. One source of truth/serialized representation for DAGs
>> (currently
>>> we
>>>>>> have SimpleDAG)
>>>>>> 
>>>>>> If we can full-fill all of these requirements and serialize at the
>>> task
>>>>>> level rather than the DAG level in the DB, then I agree that
>> probably
>>>>> makes
>>>>>> more sense.
>>>>>> 
>>>>>> 
>>>>>>> In the proposed PR’s we (Peter, Bas and me) aim to avoid
>> re-parsing
>>>> DAG
>>>>>>> files by querying all the required information from the database.
>>> In
>>>>> one
>>>>>> or
>>>>>>> two cases this may however not be possible, in which case we
>> might
>>>>> either
>>>>>>> have to fall back on the DAG file or add the missing information
>>> into
>>>>> the
>>>>>>> database. We can tackle these problems as we encounter them.
>>>>>> 
>>>>>> I think you would have the support of many of committers in
>> removing
>>>> any
>>>>>> use-cases that stand in the way of full serialization, that being
>>> said
>>>> if
>>>>>> we need to remove features we need to do this carefully and
>>>> thoughtfully,
>>>>>> and ideally with proposed alternatives/work-arounds to cover the
>>>>> removals.
>>>>>> 
>>>>>> The counter argument: this PR removes the need for the confusing
>>>>> "Refresh"
>>>>>>> button from the UI, and in general you only pay the cost for the
>>>>>> expensive
>>>>>>> DAGs when you ask about them. (I don't know what/when we call the
>>>>>>> /pickle_info endpoint of the top of my head)
>>>>>> 
>>>>>> Probably worth splitting out into a separate thread, but I'm
>> actually
>>>> not
>>>>>> sure the refresh button does anything, I think we should double
>>>> check...
>>>>> I
>>>>>> think about 2 years ago there was a commit made that made gunicorn
>>>>>> webservers automatically rotate underneath flask (each one would
>>>> reparse
>>>>>> the DAGbag). Even if it works we should probably remove it since
>> the
>>>>>> webserver refresh interval is pretty fast, and it just causes
>>> confusion
>>>>> to
>>>>>> users and implies that the DAGs are not refreshed automatically.
>>>>>> 
>>>>>> Do you mean https://json5.org/ or is this a typo? That might be
>> okay
>>>>> for a
>>>>>>> nicer user front end, but the "canonical" version stored in the
>> DB
>>>>> should
>>>>>>> be something "plainer" like just JSON.
>>>>>> 
>>>>>> I think he got this from my reply, and it was just an example, but
>>> you
>>>>> are
>>>>>> right, I agree JSON would be better than JSON5.
>>>>>> 
>>>>>> On Fri, Mar 8, 2019 at 8:53 AM Ash Berlin-Taylor <as...@apache.org>
>>>> wrote:
>>>>>> 
>>>>>>> Comments inline.
>>>>>>> 
>>>>>>>> On 8 Mar 2019, at 11:28, Kevin Yang <yr...@gmail.com> wrote:
>>>>>>>> 
>>>>>>>> Hi all,
>>>>>>>> When I was preparing some work related to this AIP I found
>>>> something
>>>>>>> very concerning. I noticed this JIRA ticket <
>>>>>>> https://issues.apache.org/jira/browse/AIRFLOW-3562> is trying to
>>>>> remove
>>>>>>> the dependency of dagbag from webserver, which is awesome--we
>>> wanted
>>>>>> badly
>>>>>>> but never got to start work on. However when I looked at some
>>>> subtasks
>>>>> of
>>>>>>> it, which try to remove dagbag dependency from each endpoint, I
>>> found
>>>>> the
>>>>>>> way we remove the dependency of dagbag is not very ideal. For
>>> example
>>>>>> this
>>>>>>> PR <https://github.com/apache/airflow/pull/4867/files> will
>>> require
>>>> us
>>>>>> to
>>>>>>> parse the dag file each time we hit the endpoint.
>>>>>>> 
>>>>>>> The counter argument: this PR removes the need for the confusing
>>>>>> "Refresh"
>>>>>>> button from the UI, and in general you only pay the cost for the
>>>>>> expensive
>>>>>>> DAGs when you ask about them. (I don't know what/when we call the
>>>>>>> /pickle_info endpoint of the top of my head)
>>>>>>> 
>>>>>>> This end point may be one to hold off on (as it can ask for
>>> multiple
>>>>>> dags)
>>>>>>> but there are some that def don't need a full dag bag or to even
>>>> parse
>>>>>> the
>>>>>>> dag file, the current DAG model has enough info.
>>>>>>> 
>>>>>>>> 
>>>>>>>> 
>>>>>>>> If we go down this path, we indeed can get rid of the dagbag
>>>>> dependency
>>>>>>> easily, but we will have to 1. increase the DB load( not too
>>>> concerning
>>>>>> at
>>>>>>> the moment ), 2. wait the DAG file to be parsed before getting
>> the
>>>> page
>>>>>>> back, potentially multiple times. DAG file can sometimes take
>>> quite a
>>>>>> while
>>>>>>> to parse, e.g. we have some framework DAG files generating large
>>>> number
>>>>>> of
>>>>>>> DAGs from some static config files or even jupyter notebooks and
>>> they
>>>>> can
>>>>>>> take 30+ seconds to parse. Yes we don't like large DAG files but
>>>> people
>>>>>> do
>>>>>>> see the beauty of code as config and sometimes heavily
>>> abuseleverage
>>>>> it.
>>>>>>> Assuming all users have the same nice small python file that can
>> be
>>>>>> parsed
>>>>>>> fast, I'm still a bit worried about this approach. Continuing on
>>> this
>>>>>> path
>>>>>>> means we've chosen DagModel to be the serialized representation
>> of
>>>> DAG
>>>>>> and
>>>>>>> DB columns to hold different properties--it can be one candidate
>>> but
>>>> I
>>>>>>> don't know if we should settle on that now. I would personally
>>>> prefer a
>>>>>>> more compact, e.g. JSON5, and easy to scale representation( such
>>> that
>>>>>>> serializing new fields != DB upgrade).
>>>>>>> 
>>>>>>> Do you mean https://json5.org/ or is this a typo? That might be
>>> okay
>>>>> for
>>>>>>> a nicer user front end, but the "canonical" version stored in the
>>> DB
>>>>>> should
>>>>>>> be something "plainer" like just JSON.
>>>>>>> 
>>>>>>> I'm not sure that "serializing new fields != DB upgrade" is that
>>> big
>>>>> of a
>>>>>>> concern, as we don't add fields that often. One possible way of
>>>> dealing
>>>>>>> with it if we do is to have a hybrid approach - a few distinct
>>>> columns,
>>>>>> but
>>>>>>> then a JSON blob. (and if we were only to support postgres we
>> could
>>>>> just
>>>>>>> use JSONb. But I think our friends at Google may object ;) )
>>>>>>> 
>>>>>>> Adding a new column in a DB migration with a default NULL
>> shouldn't
>>>> be
>>>>> an
>>>>>>> expensive operation, or difficult to achieve.
>>>>>>> 
>>>>>>> 
>>>>>>>> 
>>>>>>>> In my imagination we would have to collect the list of dynamic
>>>>> features
>>>>>>> depending on unserializable fields of a DAG and start a
>>>> discussion/vote
>>>>>> on
>>>>>>> dropping support of them( I'm working on this but if anyone has
>>>> already
>>>>>>> done so please take over), decide on the serialized
>> representation
>>>> of a
>>>>>> DAG
>>>>>>> and then replace dagbag with it in webserver. Per previous
>>> discussion
>>>>> and
>>>>>>> some offline discussions with Dan, one future of DAG
>> serialization
>>>>> that I
>>>>>>> like would look similar to this:
>>>>>>>> 
>>>>>>> 
>>>>>>>> https://imgur.com/ncqqQgc
>>>>>>> 
>>>>>>> Something I've thought about before for other things was to embed
>>> an
>>>>> API
>>>>>>> server _into_ the scheduler - this would be useful for k8s
>>>>> healthchecks,
>>>>>>> native Prometheus metrics without needed statsd bridge, and could
>>>> have
>>>>>>> endpoints to get information such as this directly.
>>>>>>> 
>>>>>>> I was thinking it would be _in_ the scheduler process using
>> either
>>>>>> threads
>>>>>>> (ick. Python's still got a GIL doesn't it?) or using
>> async/twisted
>>>> etc.
>>>>>>> (not a side-car process like we have with the logs webserver for
>>>>> `airflow
>>>>>>> worker`).
>>>>>>> 
>>>>>>> (This is possibly an unrelated discussion, but might be worth
>>> talking
>>>>>>> about?)
>>>>>>> 
>>>>>>>> We can still discuss/vote which approach we want to take but I
>>>> don't
>>>>>>> want the door to above design to be shut right now or we have to
>>>> spend
>>>>> a
>>>>>>> lot effort switch path later.
>>>>>>>> 
>>>>>>>> Bas and Peter, I'm very sorry to extend the discussion but I do
>>>> think
>>>>>>> this is tightly related to the AIP and PRs behind it. And my
>>> sincere
>>>>>>> apology for bringing this up so late( I only pull the open PR
>> list
>>>>>>> occasionally, if there's a way to subscribe to new PR event I'd
>>> love
>>>> to
>>>>>>> know how).
>>>>>>> 
>>>>>>> It's noisy, but you can subscribe to commits@airflow.apache.org
>>> (but
>>>>> be
>>>>>>> warned, this also includes all Jira tickets, edits of every
>> comment
>>>> on
>>>>>>> github etc.).
>>>>>>> 
>>>>>>> 
>>>>>>>> 
>>>>>>>> Cheers,
>>>>>>>> Kevin Y
>>>>>>>> 
>>>>>>>> On Thu, Feb 28, 2019 at 1:36 PM Peter van t Hof <
>>>>> pjrvanthof@gmail.com
>>>>>>> <ma...@gmail.com>> wrote:
>>>>>>>> Hi all,
>>>>>>>> 
>>>>>>>> Just some comments one the point Bolke dit give in relation of
>> my
>>>> PR.
>>>>>>>> 
>>>>>>>> At first, the main focus is: making the webserver stateless.
>>>>>>>> 
>>>>>>>>> 1) Make the webserver stateless: needs the graph of the
>>> *current*
>>>>> dag
>>>>>>>> 
>>>>>>>> This is the main goal but for this a lot more PR’s will be
>> coming
>>>>> once
>>>>>>> my current is merged. For edges and graph view this is covered in
>>> my
>>>> PR
>>>>>>> already.
>>>>>>>> 
>>>>>>>>> 2) Version dags: for consistency mainly and not requiring
>>> parsing
>>>>> of
>>>>>>> the
>>>>>>>>> dag on every loop
>>>>>>>> 
>>>>>>>> In my PR the historical graphs will be stored for each DagRun.
>>> This
>>>>>>> means that you can see if an older DagRun was the same graph
>>>> structure,
>>>>>>> even if some tasks does not exists anymore in the current graph.
>>>>>> Especially
>>>>>>> for dynamic DAG’s this is very useful.
>>>>>>>> 
>>>>>>>>> 3) Make the scheduler not require DAG files. This could be
>> done
>>>> if
>>>>>> the
>>>>>>>>> edges contain all information when to trigger the next task.
>> We
>>>> can
>>>>>>> then
>>>>>>>>> have event driven dag parsing outside of the scheduler loop,
>>> ie.
>>>> by
>>>>>> the
>>>>>>>>> cli. Storage can also be somewhere else (git, artifactory,
>>>>>> filesystem,
>>>>>>>>> whatever).
>>>>>>>> 
>>>>>>>> The scheduler is almost untouched in this PR. The only thing
>> that
>>>> is
>>>>>>> added is that this edges are saved to the database but the
>>> scheduling
>>>>>>> itself din’t change. The scheduler depends now still on the DAG
>>>> object.
>>>>>>>> 
>>>>>>>>> 4) Fully serialise the dag so it becomes transferable to
>>> workers
>>>>>>>> 
>>>>>>>> It nice to see that people has a lot of idea’s about this. But
>> as
>>>>> Fokko
>>>>>>> already mentioned this is out of scope for the issue what we are
>>>> trying
>>>>>> to
>>>>>>> solve. I also have some idea’s about this but I like to limit
>> this
>>>>> PR/AIP
>>>>>>> to the webserver.
>>>>>>>> 
>>>>>>>> For now my PR does solve 1 and 2 and the rest of the behaviour
>>>> (like
>>>>>>> scheduling) is untouched.
>>>>>>>> 
>>>>>>>> Gr,
>>>>>>>> Peter
>>>>>>>> 
>>>>>>> 
>>>>>>> 
>>>>>> 
>>>>> 
>>>> 
>>> 
>> 

Re: [DISCUSS] AIP-12 Persist DAG into DB

[Tao Feng <fe...@gmail.com>]

Do we reach any consensus on this topic /AIP? I think persisting DAG is
pretty important actually.

-Tao

On Tue, Mar 12, 2019 at 3:01 AM Kevin Yang <yr...@gmail.com> wrote:

> Hi Fokko,
>
> As a large cluster maintainer, I’m not a big fan of large DAG files
> neither. But I’m not sure if I’ll consider this bad practice. We have some
> large frameworks, e.g. experimentation and machine learning, that are
> complex by nature and generate large number of DAGs from their customer
> configs to get better flexibility. I consider them as advance use cases of
> Airflow and open up a lot potentials for Airflow, unless we’ve previously
> set some boundaries around how complex DAG codes can be that I’m not aware
> of. About resulting in an unworkable situation, yes we are experiencing
> pain from having such large DAG files, mainly on the webserver side, but
> the system overall are running stable. We are actually hoping to improve
> the situation by applying solutions like making webserver stateless. It is
> ok that if the owners of large DAG files need to pay but we should try
> minimize the price—longer refresh interval, extra task running time, but
> nothing too crazy.
>
>
> I think we’re aligned on storing info in DB as long as we can meet the
> requirements Dan mentioned earlier—we just need that balance decided, so
> I’m gonna skip this part( out of all the requirements, No.1 seems to be
> least clear, maybe we can expand on that). One thing about the proposed
> idea is that we implicitly couple DagRun with DAG version, which at the
> first glance make sense but imo not very ideal. I feel full versioning
> should track all changes instead of tracking changes only when we create
> DagRun. E.g. my task failed and I merged new code to fix my task and I want
> to rerun it with the current code, with serialize DAG during DagRun
> creation time we won’t have the up to date snapshot—sure we can work around
> it by like always keep a current snapshot of DAG but this is kinda messy
> and confusing. This is what popped up on the top of my head and w/o full
> versioning we might have some other tricky cases, e.g. ur backfill case.
> But I just gave a few thoughts into this and you might already have a
> complete story that will void my concerns.
>
>
> Cheers,
> Kevin Y
>
> On Sun, Mar 10, 2019 at 11:29 AM Driesprong, Fokko <fo...@driesprong.frl>
> wrote:
>
> > Thanks Kevin for opening the discussion. I think it is important to have
> a
> > clear overview on how to approach the AIP.
> >
> > First of all, how many DAGs do we have that take 30s to parse? I consider
> > this bad practice, and this would also result in an unworkable situation
> > with the current setup of Airflow since it will take a lot of resources
> on
> > the webserver/scheduler, and the whole system will become unresponsive. I
> > will be hard to cope with such DAGs in general.
> >
> > The idea from the AIP is to have the versioned version of the dag in the
> > DB, so in the end, you won't need to parse the whole thing every time.
> Only
> > when you trigger a DAG, or when you want to see the current status of the
> > dag.
> >
> > Like stated earlier, I strongly feel we shouldn't serialize the DAGs as
> > JSON(5) or pickles in general. For me, this is deferring the pain of
> > setting up a structure of the DAG object itself.
> > Having the DAG denormalized in the database will give us cleaner storage
> of
> > our DAG. We can, for example, enforce fields by making them not null, so
> we
> > know that is something is off at write time, instead of read.
> Furthermore,
> > we're missing logical types such as dates, which we efficiently can query
> > using the indices of the database.
> > Also, with all serialization formats, evolution isn't trivial. Consider
> the
> > situations when:
> > - We're introducing a new field, and it might be null, therefore we need
> to
> > bake in all kinds of logic into the Airflow code, which you don't want.
> > With proper migration scripts, you could prefill these fields, and make
> > them not null.
> > - Changing the models, for example, you still can't change a string-type
> > into a integer with adding custom logic. In this case, the reviewer needs
> > to be extra careful that there are no breaking changes introduced. Right
> > now we're doing minimal forward- and backward compatibilitytesting.
> >
> > In the case we get too many migrations, we could also squash (some of
> them)
> > when preparing the release.
> >
> > Personally, I don't think the serialization is the issue here. As Max
> > already mentioned, it is the optimal balance of (de)normalization. From
> the
> > user perspective, the serialization won't change much of the behaviour of
> > Airflow.
> >
> > For me, instead of having `DAG.serialize()` and `DAG.deser(version)` is
> not
> > the ideal approach. But it might be that we're on the same page :-) I
> > believe it should be something like `DagRun.find('fokkos_dag',
> > datetime(2018, 03, 01))` and construct the correct version of the dag.
> > Since there is an uniqueness constrain on dag_id, datetime, this will
> > always return the same dag. You will get the versioned DagRun as it ran
> > that time. Serializing the fields adn storing them in the database should
> > happen transparently when you update the DAG object. When you run a dag,
> > you'll parse the dag, and then run it. `Dag().create_dagrun(...)`, this
> > will create a DagRun as the name suggests, if the version of the dag
> still
> > exists in the database, it will reuse that one, otherwise it will create
> a
> > new version of the DAG (with all the operators etc). In this sense the
> > version of the DAGs should be done within the Dag(Run).
> >
> > The versioning will change the behavour from a user perspective. Right
> now
> > we store only a single version. For example, the poor mans backfilling
> > won't work anymore. This is clearing the state from past&future, up- and
> > downstream, and let it catch up again.
> > In this case, the old version of the DAG won't exists anymore, and
> > potentially there are tasks that aren't in the code anymore. In this case
> > we need to clear the version of the dag, and rerun it with the latest
> > version `DagRun.find('fokkos_dag', datetime(2018, 03, 01)).clear()`. How
> we
> > are going to do clear's downstram in the middle of the dag, that is
> > something I still have to figure out. Because potentially there are tasks
> > that can't be rerun because the underlying Python code has changed, both
> on
> > user level as on Airflow level. It will be impossible to get these
> features
> > pure in that sense.
> > I would not suggest adding a new status in here, indicating that the task
> > can't be rerun since it isn't part of the DAG anymore. We have to find
> the
> > balance here in adding complexity (also to the scheduler) and features
> that
> > we need to introduce to help the user.
> >
> > Cheers, Fokko
> >
> > Ps. Jarek, interesting idea. It shouldn't be too hard to make Airflow
> more
> > k8s native. You could package your dags within your container, and do a
> > rolling update. Add the DAGs as the last layer, and then point the DAGs
> > folder to the same location. The hard part here is that you need to
> > gracefuly restart the workers. Currently AFAIK the signals given to the
> pod
> > aren't respected. So when the scheduler/webserver/worker receives a
> > SIGTERM, it should stop the jobs nicely and then exit the container,
> before
> > k8s kills the container using a SIGKILL.  This will be challenging with
> the
> > workers, which they are potentially long-running. Maybe stop kicking off
> > new jobs, and let the old ones finish, will be good enough, but then we
> > need to increase the standard kill timeout substantially. Having this
> would
> > also enable the autoscaling of the workers.
> >
> >
> >
> > Op za 9 mrt. 2019 om 19:07 schreef Maxime Beauchemin <
> > maximebeauchemin@gmail.com>:
> >
> > > I want to raise the question of the amount of normalization we want to
> > use
> > > here as it seems the to be an area that needs more attention.
> > >
> > > The SIP suggest having DAG blobs, task blobs and edges (call it the
> > > fairly-normalized). I also like the idea of all-encompassing (call it
> > > very-denormalized) DAG blobs as it seems easier to manage in terms of
> > > versioning. The question here is whether we go with one of these method
> > > exclusively, something in-between or even a hybrid approach (redundant
> > > blobs that use different level of normalization).
> > >
> > > It's nice and simple to just push or pull DAG atomic objects with a
> > version
> > > stamp on it. It's clearly simpler than dealing with 3 versioned tables
> > > (dag, tasks, edges). There are a lot of pros/cons, and they become more
> > > apparent with the perspective of very large DAGs. If the web server is
> > > building a "task details page", using the "fairly-normalized" model, it
> > can
> > > just pull what it needs instead of pulling the large DAG blob.
> Similarly,
> > > if building a sub-tree view (a subset of the DAG), perhaps it can only
> > > retrieve what it needs. But if you need the whole DAG (say for the
> > > scheduler use case) then you're dealing with more complex SQL/ORM
> > > operations (joins hopefully, or multiple db round trips)
> > >
> > > Now maybe the right approach is more something like 2 tables: DAG and
> > > task_details, where edges keys are denormalized into DAG (arguably
> > that's a
> > > few KBs at most, even for large DAGs), and maybe the DAG object has
> most
> > of
> > > the high level task metadata information (operator, name, baseoperator
> > key
> > > attrs), and task_details has the big blobs (SQL code). This is
> probably a
> > > nice compromise, the question becomes "how much task-level detail do we
> > > store in the DAG-centric blog?", probably not much to keep the DAG
> > objects
> > > as small as possible. The main downside here is that you cannot have
> the
> > > database join and have to do 2 round trips to reconstruct a DAG object
> > > (fetch the DAG, parse the object to get the list of tasks, and then run
> > > another db query to get those task details).
> > >
> > > To resume, I'd qualify the more normalized approach as the most proper,
> > but
> > > also the more complex. It'll shine in specific cases around large DAGs.
> > If
> > > we have the proper abstractions (methods like DAG.serialize(),
> > > DAG.deser(version)) then I guess that's not an issue.
> > >
> > > Max
> > >
> > > On Fri, Mar 8, 2019 at 5:21 PM Kevin Yang <yr...@gmail.com> wrote:
> > >
> > > > Hi Julian, I'm definitely aligned with you guys on making the
> webserver
> > > > independent of DAG parsing, just the end goal to me would be to
> build a
> > > > complete story around serializing DAG--and move with the story in
> > mind. I
> > > > feel like you guys may already have a list of dynamic features we
> need
> > to
> > > > deprecate/change, if that is the case feel free to open the
> discussion
> > on
> > > > what we do to them with DAG serialization.
> > > >
> > > > Julian, Ash, Dan, on 2nd thought I do agree that if we can meet the
> > > > requirements Dan mentioned, it would be nice to have them stored in
> the
> > > DB.
> > > > Some combined solutions like having a column of serialized graph in
> the
> > > > serialized dag table can potentially meet all requirements. What
> format
> > > we
> > > > end up using to represent DAG between components is now less
> important
> > > > IMO--fine to refactor those endpoints only need DagModel to use only
> > > > DagModel, easy to do a batch replacement if we decide otherwise
> later.
> > > More
> > > > important is to define this source of truth for serialized DAG.
> > > >
> > > > Ash, ty for the email list, I'll tune my filters accordingly :D I'm
> > > leaning
> > > > towards having a separate process for the parser so we got no
> scheduler
> > > > dependency etc for this parser but we can discuss this in another
> > thread.
> > > >
> > > > On Fri, Mar 8, 2019 at 8:57 AM Dan Davydov
> > <ddavydov@twitter.com.invalid
> > > >
> > > > wrote:
> > > >
> > > > > >
> > > > > > Personally I don’t understand why people are pushing for a
> > JSON-based
> > > > DAG
> > > > > > representation
> > > > >
> > > > > It sounds like you agree that DAGs should be serialized (just in
> the
> > DB
> > > > > instead of JSON), so will only address why JSON is better than
> MySQL
> > > (AKA
> > > > > serializing at the DAG level vs the task level) as far as I can
> see,
> > > and
> > > > > not why we need serialization. If you zoom out and look at all the
> > use
> > > > > cases of serialized DAGs, e.g. having the scheduler use them
> instead
> > of
> > > > > parsing DAGs directly, then it becomes clear that we need all
> > > appropriate
> > > > > metadata in these DAGs, (operator params, DAG properties, etc), in
> > > which
> > > > > case it's not clear how it will fit nicely into a DB table (unless
> > you
> > > > > wanted to do something like (parent_task_id, task_id, task_params),
> > > also
> > > > > keep in mind that we will need to store different versions of each
> > DAG
> > > in
> > > > > the future so that we can ensure consistency in a dagrun, i.e. each
> > > task
> > > > in
> > > > > a dagrun uses the same version of a DAG.
> > > > >
> > > > > I think some of our requirements should be:
> > > > > 1. The data model will lead to acceptable performance in all of its
> > > > > consumers (scheduler, webserver, workers), i.e. no n+1 access
> > patterns
> > > > (my
> > > > > biggest concern about serializing at task level as you propose vs
> at
> > > DAG
> > > > > level)
> > > > > 2. We can have versioning of serialized DAGs
> > > > > 3. The ability to separate DAGs into their own data store (e.g. no
> > > > reliance
> > > > > on joins between the new table and the old one)
> > > > > 4. One source of truth/serialized representation for DAGs
> (currently
> > we
> > > > > have SimpleDAG)
> > > > >
> > > > > If we can full-fill all of these requirements and serialize at the
> > task
> > > > > level rather than the DAG level in the DB, then I agree that
> probably
> > > > makes
> > > > > more sense.
> > > > >
> > > > >
> > > > > > In the proposed PR’s we (Peter, Bas and me) aim to avoid
> re-parsing
> > > DAG
> > > > > > files by querying all the required information from the database.
> > In
> > > > one
> > > > > or
> > > > > > two cases this may however not be possible, in which case we
> might
> > > > either
> > > > > > have to fall back on the DAG file or add the missing information
> > into
> > > > the
> > > > > > database. We can tackle these problems as we encounter them.
> > > > >
> > > > > I think you would have the support of many of committers in
> removing
> > > any
> > > > > use-cases that stand in the way of full serialization, that being
> > said
> > > if
> > > > > we need to remove features we need to do this carefully and
> > > thoughtfully,
> > > > > and ideally with proposed alternatives/work-arounds to cover the
> > > > removals.
> > > > >
> > > > > The counter argument: this PR removes the need for the confusing
> > > > "Refresh"
> > > > > > button from the UI, and in general you only pay the cost for the
> > > > > expensive
> > > > > > DAGs when you ask about them. (I don't know what/when we call the
> > > > > > /pickle_info endpoint of the top of my head)
> > > > >
> > > > > Probably worth splitting out into a separate thread, but I'm
> actually
> > > not
> > > > > sure the refresh button does anything, I think we should double
> > > check...
> > > > I
> > > > > think about 2 years ago there was a commit made that made gunicorn
> > > > > webservers automatically rotate underneath flask (each one would
> > > reparse
> > > > > the DAGbag). Even if it works we should probably remove it since
> the
> > > > > webserver refresh interval is pretty fast, and it just causes
> > confusion
> > > > to
> > > > > users and implies that the DAGs are not refreshed automatically.
> > > > >
> > > > > Do you mean https://json5.org/ or is this a typo? That might be
> okay
> > > > for a
> > > > > > nicer user front end, but the "canonical" version stored in the
> DB
> > > > should
> > > > > > be something "plainer" like just JSON.
> > > > >
> > > > > I think he got this from my reply, and it was just an example, but
> > you
> > > > are
> > > > > right, I agree JSON would be better than JSON5.
> > > > >
> > > > > On Fri, Mar 8, 2019 at 8:53 AM Ash Berlin-Taylor <as...@apache.org>
> > > wrote:
> > > > >
> > > > > > Comments inline.
> > > > > >
> > > > > > > On 8 Mar 2019, at 11:28, Kevin Yang <yr...@gmail.com> wrote:
> > > > > > >
> > > > > > > Hi all,
> > > > > > > When I was preparing some work related to this AIP I found
> > > something
> > > > > > very concerning. I noticed this JIRA ticket <
> > > > > > https://issues.apache.org/jira/browse/AIRFLOW-3562> is trying to
> > > > remove
> > > > > > the dependency of dagbag from webserver, which is awesome--we
> > wanted
> > > > > badly
> > > > > > but never got to start work on. However when I looked at some
> > > subtasks
> > > > of
> > > > > > it, which try to remove dagbag dependency from each endpoint, I
> > found
> > > > the
> > > > > > way we remove the dependency of dagbag is not very ideal. For
> > example
> > > > > this
> > > > > > PR <https://github.com/apache/airflow/pull/4867/files> will
> > require
> > > us
> > > > > to
> > > > > > parse the dag file each time we hit the endpoint.
> > > > > >
> > > > > > The counter argument: this PR removes the need for the confusing
> > > > > "Refresh"
> > > > > > button from the UI, and in general you only pay the cost for the
> > > > > expensive
> > > > > > DAGs when you ask about them. (I don't know what/when we call the
> > > > > > /pickle_info endpoint of the top of my head)
> > > > > >
> > > > > > This end point may be one to hold off on (as it can ask for
> > multiple
> > > > > dags)
> > > > > > but there are some that def don't need a full dag bag or to even
> > > parse
> > > > > the
> > > > > > dag file, the current DAG model has enough info.
> > > > > >
> > > > > > >
> > > > > > >
> > > > > > > If we go down this path, we indeed can get rid of the dagbag
> > > > dependency
> > > > > > easily, but we will have to 1. increase the DB load( not too
> > > concerning
> > > > > at
> > > > > > the moment ), 2. wait the DAG file to be parsed before getting
> the
> > > page
> > > > > > back, potentially multiple times. DAG file can sometimes take
> > quite a
> > > > > while
> > > > > > to parse, e.g. we have some framework DAG files generating large
> > > number
> > > > > of
> > > > > > DAGs from some static config files or even jupyter notebooks and
> > they
> > > > can
> > > > > > take 30+ seconds to parse. Yes we don't like large DAG files but
> > > people
> > > > > do
> > > > > > see the beauty of code as config and sometimes heavily
> > abuseleverage
> > > > it.
> > > > > > Assuming all users have the same nice small python file that can
> be
> > > > > parsed
> > > > > > fast, I'm still a bit worried about this approach. Continuing on
> > this
> > > > > path
> > > > > > means we've chosen DagModel to be the serialized representation
> of
> > > DAG
> > > > > and
> > > > > > DB columns to hold different properties--it can be one candidate
> > but
> > > I
> > > > > > don't know if we should settle on that now. I would personally
> > > prefer a
> > > > > > more compact, e.g. JSON5, and easy to scale representation( such
> > that
> > > > > > serializing new fields != DB upgrade).
> > > > > >
> > > > > > Do you mean https://json5.org/ or is this a typo? That might be
> > okay
> > > > for
> > > > > > a nicer user front end, but the "canonical" version stored in the
> > DB
> > > > > should
> > > > > > be something "plainer" like just JSON.
> > > > > >
> > > > > > I'm not sure that "serializing new fields != DB upgrade" is that
> > big
> > > > of a
> > > > > > concern, as we don't add fields that often. One possible way of
> > > dealing
> > > > > > with it if we do is to have a hybrid approach - a few distinct
> > > columns,
> > > > > but
> > > > > > then a JSON blob. (and if we were only to support postgres we
> could
> > > > just
> > > > > > use JSONb. But I think our friends at Google may object ;) )
> > > > > >
> > > > > > Adding a new column in a DB migration with a default NULL
> shouldn't
> > > be
> > > > an
> > > > > > expensive operation, or difficult to achieve.
> > > > > >
> > > > > >
> > > > > > >
> > > > > > > In my imagination we would have to collect the list of dynamic
> > > > features
> > > > > > depending on unserializable fields of a DAG and start a
> > > discussion/vote
> > > > > on
> > > > > > dropping support of them( I'm working on this but if anyone has
> > > already
> > > > > > done so please take over), decide on the serialized
> representation
> > > of a
> > > > > DAG
> > > > > > and then replace dagbag with it in webserver. Per previous
> > discussion
> > > > and
> > > > > > some offline discussions with Dan, one future of DAG
> serialization
> > > > that I
> > > > > > like would look similar to this:
> > > > > > >
> > > > > >
> > > > > > > https://imgur.com/ncqqQgc
> > > > > >
> > > > > > Something I've thought about before for other things was to embed
> > an
> > > > API
> > > > > > server _into_ the scheduler - this would be useful for k8s
> > > > healthchecks,
> > > > > > native Prometheus metrics without needed statsd bridge, and could
> > > have
> > > > > > endpoints to get information such as this directly.
> > > > > >
> > > > > > I was thinking it would be _in_ the scheduler process using
> either
> > > > > threads
> > > > > > (ick. Python's still got a GIL doesn't it?) or using
> async/twisted
> > > etc.
> > > > > > (not a side-car process like we have with the logs webserver for
> > > > `airflow
> > > > > > worker`).
> > > > > >
> > > > > > (This is possibly an unrelated discussion, but might be worth
> > talking
> > > > > > about?)
> > > > > >
> > > > > > > We can still discuss/vote which approach we want to take but I
> > > don't
> > > > > > want the door to above design to be shut right now or we have to
> > > spend
> > > > a
> > > > > > lot effort switch path later.
> > > > > > >
> > > > > > > Bas and Peter, I'm very sorry to extend the discussion but I do
> > > think
> > > > > > this is tightly related to the AIP and PRs behind it. And my
> > sincere
> > > > > > apology for bringing this up so late( I only pull the open PR
> list
> > > > > > occasionally, if there's a way to subscribe to new PR event I'd
> > love
> > > to
> > > > > > know how).
> > > > > >
> > > > > > It's noisy, but you can subscribe to commits@airflow.apache.org
> > (but
> > > > be
> > > > > > warned, this also includes all Jira tickets, edits of every
> comment
> > > on
> > > > > > github etc.).
> > > > > >
> > > > > >
> > > > > > >
> > > > > > > Cheers,
> > > > > > > Kevin Y
> > > > > > >
> > > > > > > On Thu, Feb 28, 2019 at 1:36 PM Peter van t Hof <
> > > > pjrvanthof@gmail.com
> > > > > > <ma...@gmail.com>> wrote:
> > > > > > > Hi all,
> > > > > > >
> > > > > > > Just some comments one the point Bolke dit give in relation of
> my
> > > PR.
> > > > > > >
> > > > > > > At first, the main focus is: making the webserver stateless.
> > > > > > >
> > > > > > > > 1) Make the webserver stateless: needs the graph of the
> > *current*
> > > > dag
> > > > > > >
> > > > > > > This is the main goal but for this a lot more PR’s will be
> coming
> > > > once
> > > > > > my current is merged. For edges and graph view this is covered in
> > my
> > > PR
> > > > > > already.
> > > > > > >
> > > > > > > > 2) Version dags: for consistency mainly and not requiring
> > parsing
> > > > of
> > > > > > the
> > > > > > > > dag on every loop
> > > > > > >
> > > > > > > In my PR the historical graphs will be stored for each DagRun.
> > This
> > > > > > means that you can see if an older DagRun was the same graph
> > > structure,
> > > > > > even if some tasks does not exists anymore in the current graph.
> > > > > Especially
> > > > > > for dynamic DAG’s this is very useful.
> > > > > > >
> > > > > > > > 3) Make the scheduler not require DAG files. This could be
> done
> > > if
> > > > > the
> > > > > > > > edges contain all information when to trigger the next task.
> We
> > > can
> > > > > > then
> > > > > > > > have event driven dag parsing outside of the scheduler loop,
> > ie.
> > > by
> > > > > the
> > > > > > > > cli. Storage can also be somewhere else (git, artifactory,
> > > > > filesystem,
> > > > > > > > whatever).
> > > > > > >
> > > > > > > The scheduler is almost untouched in this PR. The only thing
> that
> > > is
> > > > > > added is that this edges are saved to the database but the
> > scheduling
> > > > > > itself din’t change. The scheduler depends now still on the DAG
> > > object.
> > > > > > >
> > > > > > > > 4) Fully serialise the dag so it becomes transferable to
> > workers
> > > > > > >
> > > > > > > It nice to see that people has a lot of idea’s about this. But
> as
> > > > Fokko
> > > > > > already mentioned this is out of scope for the issue what we are
> > > trying
> > > > > to
> > > > > > solve. I also have some idea’s about this but I like to limit
> this
> > > > PR/AIP
> > > > > > to the webserver.
> > > > > > >
> > > > > > > For now my PR does solve 1 and 2 and the rest of the behaviour
> > > (like
> > > > > > scheduling) is untouched.
> > > > > > >
> > > > > > > Gr,
> > > > > > > Peter
> > > > > > >
> > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
>

Re: [DISCUSS] AIP-12 Persist DAG into DB

[Kevin Yang <yr...@gmail.com>]

Hi Fokko,

As a large cluster maintainer, I’m not a big fan of large DAG files
neither. But I’m not sure if I’ll consider this bad practice. We have some
large frameworks, e.g. experimentation and machine learning, that are
complex by nature and generate large number of DAGs from their customer
configs to get better flexibility. I consider them as advance use cases of
Airflow and open up a lot potentials for Airflow, unless we’ve previously
set some boundaries around how complex DAG codes can be that I’m not aware
of. About resulting in an unworkable situation, yes we are experiencing
pain from having such large DAG files, mainly on the webserver side, but
the system overall are running stable. We are actually hoping to improve
the situation by applying solutions like making webserver stateless. It is
ok that if the owners of large DAG files need to pay but we should try
minimize the price—longer refresh interval, extra task running time, but
nothing too crazy.


I think we’re aligned on storing info in DB as long as we can meet the
requirements Dan mentioned earlier—we just need that balance decided, so
I’m gonna skip this part( out of all the requirements, No.1 seems to be
least clear, maybe we can expand on that). One thing about the proposed
idea is that we implicitly couple DagRun with DAG version, which at the
first glance make sense but imo not very ideal. I feel full versioning
should track all changes instead of tracking changes only when we create
DagRun. E.g. my task failed and I merged new code to fix my task and I want
to rerun it with the current code, with serialize DAG during DagRun
creation time we won’t have the up to date snapshot—sure we can work around
it by like always keep a current snapshot of DAG but this is kinda messy
and confusing. This is what popped up on the top of my head and w/o full
versioning we might have some other tricky cases, e.g. ur backfill case.
But I just gave a few thoughts into this and you might already have a
complete story that will void my concerns.


Cheers,
Kevin Y

On Sun, Mar 10, 2019 at 11:29 AM Driesprong, Fokko <fo...@driesprong.frl>
wrote:

> Thanks Kevin for opening the discussion. I think it is important to have a
> clear overview on how to approach the AIP.
>
> First of all, how many DAGs do we have that take 30s to parse? I consider
> this bad practice, and this would also result in an unworkable situation
> with the current setup of Airflow since it will take a lot of resources on
> the webserver/scheduler, and the whole system will become unresponsive. I
> will be hard to cope with such DAGs in general.
>
> The idea from the AIP is to have the versioned version of the dag in the
> DB, so in the end, you won't need to parse the whole thing every time. Only
> when you trigger a DAG, or when you want to see the current status of the
> dag.
>
> Like stated earlier, I strongly feel we shouldn't serialize the DAGs as
> JSON(5) or pickles in general. For me, this is deferring the pain of
> setting up a structure of the DAG object itself.
> Having the DAG denormalized in the database will give us cleaner storage of
> our DAG. We can, for example, enforce fields by making them not null, so we
> know that is something is off at write time, instead of read. Furthermore,
> we're missing logical types such as dates, which we efficiently can query
> using the indices of the database.
> Also, with all serialization formats, evolution isn't trivial. Consider the
> situations when:
> - We're introducing a new field, and it might be null, therefore we need to
> bake in all kinds of logic into the Airflow code, which you don't want.
> With proper migration scripts, you could prefill these fields, and make
> them not null.
> - Changing the models, for example, you still can't change a string-type
> into a integer with adding custom logic. In this case, the reviewer needs
> to be extra careful that there are no breaking changes introduced. Right
> now we're doing minimal forward- and backward compatibilitytesting.
>
> In the case we get too many migrations, we could also squash (some of them)
> when preparing the release.
>
> Personally, I don't think the serialization is the issue here. As Max
> already mentioned, it is the optimal balance of (de)normalization. From the
> user perspective, the serialization won't change much of the behaviour of
> Airflow.
>
> For me, instead of having `DAG.serialize()` and `DAG.deser(version)` is not
> the ideal approach. But it might be that we're on the same page :-) I
> believe it should be something like `DagRun.find('fokkos_dag',
> datetime(2018, 03, 01))` and construct the correct version of the dag.
> Since there is an uniqueness constrain on dag_id, datetime, this will
> always return the same dag. You will get the versioned DagRun as it ran
> that time. Serializing the fields adn storing them in the database should
> happen transparently when you update the DAG object. When you run a dag,
> you'll parse the dag, and then run it. `Dag().create_dagrun(...)`, this
> will create a DagRun as the name suggests, if the version of the dag still
> exists in the database, it will reuse that one, otherwise it will create a
> new version of the DAG (with all the operators etc). In this sense the
> version of the DAGs should be done within the Dag(Run).
>
> The versioning will change the behavour from a user perspective. Right now
> we store only a single version. For example, the poor mans backfilling
> won't work anymore. This is clearing the state from past&future, up- and
> downstream, and let it catch up again.
> In this case, the old version of the DAG won't exists anymore, and
> potentially there are tasks that aren't in the code anymore. In this case
> we need to clear the version of the dag, and rerun it with the latest
> version `DagRun.find('fokkos_dag', datetime(2018, 03, 01)).clear()`. How we
> are going to do clear's downstram in the middle of the dag, that is
> something I still have to figure out. Because potentially there are tasks
> that can't be rerun because the underlying Python code has changed, both on
> user level as on Airflow level. It will be impossible to get these features
> pure in that sense.
> I would not suggest adding a new status in here, indicating that the task
> can't be rerun since it isn't part of the DAG anymore. We have to find the
> balance here in adding complexity (also to the scheduler) and features that
> we need to introduce to help the user.
>
> Cheers, Fokko
>
> Ps. Jarek, interesting idea. It shouldn't be too hard to make Airflow more
> k8s native. You could package your dags within your container, and do a
> rolling update. Add the DAGs as the last layer, and then point the DAGs
> folder to the same location. The hard part here is that you need to
> gracefuly restart the workers. Currently AFAIK the signals given to the pod
> aren't respected. So when the scheduler/webserver/worker receives a
> SIGTERM, it should stop the jobs nicely and then exit the container, before
> k8s kills the container using a SIGKILL.  This will be challenging with the
> workers, which they are potentially long-running. Maybe stop kicking off
> new jobs, and let the old ones finish, will be good enough, but then we
> need to increase the standard kill timeout substantially. Having this would
> also enable the autoscaling of the workers.
>
>
>
> Op za 9 mrt. 2019 om 19:07 schreef Maxime Beauchemin <
> maximebeauchemin@gmail.com>:
>
> > I want to raise the question of the amount of normalization we want to
> use
> > here as it seems the to be an area that needs more attention.
> >
> > The SIP suggest having DAG blobs, task blobs and edges (call it the
> > fairly-normalized). I also like the idea of all-encompassing (call it
> > very-denormalized) DAG blobs as it seems easier to manage in terms of
> > versioning. The question here is whether we go with one of these method
> > exclusively, something in-between or even a hybrid approach (redundant
> > blobs that use different level of normalization).
> >
> > It's nice and simple to just push or pull DAG atomic objects with a
> version
> > stamp on it. It's clearly simpler than dealing with 3 versioned tables
> > (dag, tasks, edges). There are a lot of pros/cons, and they become more
> > apparent with the perspective of very large DAGs. If the web server is
> > building a "task details page", using the "fairly-normalized" model, it
> can
> > just pull what it needs instead of pulling the large DAG blob. Similarly,
> > if building a sub-tree view (a subset of the DAG), perhaps it can only
> > retrieve what it needs. But if you need the whole DAG (say for the
> > scheduler use case) then you're dealing with more complex SQL/ORM
> > operations (joins hopefully, or multiple db round trips)
> >
> > Now maybe the right approach is more something like 2 tables: DAG and
> > task_details, where edges keys are denormalized into DAG (arguably
> that's a
> > few KBs at most, even for large DAGs), and maybe the DAG object has most
> of
> > the high level task metadata information (operator, name, baseoperator
> key
> > attrs), and task_details has the big blobs (SQL code). This is probably a
> > nice compromise, the question becomes "how much task-level detail do we
> > store in the DAG-centric blog?", probably not much to keep the DAG
> objects
> > as small as possible. The main downside here is that you cannot have the
> > database join and have to do 2 round trips to reconstruct a DAG object
> > (fetch the DAG, parse the object to get the list of tasks, and then run
> > another db query to get those task details).
> >
> > To resume, I'd qualify the more normalized approach as the most proper,
> but
> > also the more complex. It'll shine in specific cases around large DAGs.
> If
> > we have the proper abstractions (methods like DAG.serialize(),
> > DAG.deser(version)) then I guess that's not an issue.
> >
> > Max
> >
> > On Fri, Mar 8, 2019 at 5:21 PM Kevin Yang <yr...@gmail.com> wrote:
> >
> > > Hi Julian, I'm definitely aligned with you guys on making the webserver
> > > independent of DAG parsing, just the end goal to me would be to build a
> > > complete story around serializing DAG--and move with the story in
> mind. I
> > > feel like you guys may already have a list of dynamic features we need
> to
> > > deprecate/change, if that is the case feel free to open the discussion
> on
> > > what we do to them with DAG serialization.
> > >
> > > Julian, Ash, Dan, on 2nd thought I do agree that if we can meet the
> > > requirements Dan mentioned, it would be nice to have them stored in the
> > DB.
> > > Some combined solutions like having a column of serialized graph in the
> > > serialized dag table can potentially meet all requirements. What format
> > we
> > > end up using to represent DAG between components is now less important
> > > IMO--fine to refactor those endpoints only need DagModel to use only
> > > DagModel, easy to do a batch replacement if we decide otherwise later.
> > More
> > > important is to define this source of truth for serialized DAG.
> > >
> > > Ash, ty for the email list, I'll tune my filters accordingly :D I'm
> > leaning
> > > towards having a separate process for the parser so we got no scheduler
> > > dependency etc for this parser but we can discuss this in another
> thread.
> > >
> > > On Fri, Mar 8, 2019 at 8:57 AM Dan Davydov
> <ddavydov@twitter.com.invalid
> > >
> > > wrote:
> > >
> > > > >
> > > > > Personally I don’t understand why people are pushing for a
> JSON-based
> > > DAG
> > > > > representation
> > > >
> > > > It sounds like you agree that DAGs should be serialized (just in the
> DB
> > > > instead of JSON), so will only address why JSON is better than MySQL
> > (AKA
> > > > serializing at the DAG level vs the task level) as far as I can see,
> > and
> > > > not why we need serialization. If you zoom out and look at all the
> use
> > > > cases of serialized DAGs, e.g. having the scheduler use them instead
> of
> > > > parsing DAGs directly, then it becomes clear that we need all
> > appropriate
> > > > metadata in these DAGs, (operator params, DAG properties, etc), in
> > which
> > > > case it's not clear how it will fit nicely into a DB table (unless
> you
> > > > wanted to do something like (parent_task_id, task_id, task_params),
> > also
> > > > keep in mind that we will need to store different versions of each
> DAG
> > in
> > > > the future so that we can ensure consistency in a dagrun, i.e. each
> > task
> > > in
> > > > a dagrun uses the same version of a DAG.
> > > >
> > > > I think some of our requirements should be:
> > > > 1. The data model will lead to acceptable performance in all of its
> > > > consumers (scheduler, webserver, workers), i.e. no n+1 access
> patterns
> > > (my
> > > > biggest concern about serializing at task level as you propose vs at
> > DAG
> > > > level)
> > > > 2. We can have versioning of serialized DAGs
> > > > 3. The ability to separate DAGs into their own data store (e.g. no
> > > reliance
> > > > on joins between the new table and the old one)
> > > > 4. One source of truth/serialized representation for DAGs (currently
> we
> > > > have SimpleDAG)
> > > >
> > > > If we can full-fill all of these requirements and serialize at the
> task
> > > > level rather than the DAG level in the DB, then I agree that probably
> > > makes
> > > > more sense.
> > > >
> > > >
> > > > > In the proposed PR’s we (Peter, Bas and me) aim to avoid re-parsing
> > DAG
> > > > > files by querying all the required information from the database.
> In
> > > one
> > > > or
> > > > > two cases this may however not be possible, in which case we might
> > > either
> > > > > have to fall back on the DAG file or add the missing information
> into
> > > the
> > > > > database. We can tackle these problems as we encounter them.
> > > >
> > > > I think you would have the support of many of committers in removing
> > any
> > > > use-cases that stand in the way of full serialization, that being
> said
> > if
> > > > we need to remove features we need to do this carefully and
> > thoughtfully,
> > > > and ideally with proposed alternatives/work-arounds to cover the
> > > removals.
> > > >
> > > > The counter argument: this PR removes the need for the confusing
> > > "Refresh"
> > > > > button from the UI, and in general you only pay the cost for the
> > > > expensive
> > > > > DAGs when you ask about them. (I don't know what/when we call the
> > > > > /pickle_info endpoint of the top of my head)
> > > >
> > > > Probably worth splitting out into a separate thread, but I'm actually
> > not
> > > > sure the refresh button does anything, I think we should double
> > check...
> > > I
> > > > think about 2 years ago there was a commit made that made gunicorn
> > > > webservers automatically rotate underneath flask (each one would
> > reparse
> > > > the DAGbag). Even if it works we should probably remove it since the
> > > > webserver refresh interval is pretty fast, and it just causes
> confusion
> > > to
> > > > users and implies that the DAGs are not refreshed automatically.
> > > >
> > > > Do you mean https://json5.org/ or is this a typo? That might be okay
> > > for a
> > > > > nicer user front end, but the "canonical" version stored in the DB
> > > should
> > > > > be something "plainer" like just JSON.
> > > >
> > > > I think he got this from my reply, and it was just an example, but
> you
> > > are
> > > > right, I agree JSON would be better than JSON5.
> > > >
> > > > On Fri, Mar 8, 2019 at 8:53 AM Ash Berlin-Taylor <as...@apache.org>
> > wrote:
> > > >
> > > > > Comments inline.
> > > > >
> > > > > > On 8 Mar 2019, at 11:28, Kevin Yang <yr...@gmail.com> wrote:
> > > > > >
> > > > > > Hi all,
> > > > > > When I was preparing some work related to this AIP I found
> > something
> > > > > very concerning. I noticed this JIRA ticket <
> > > > > https://issues.apache.org/jira/browse/AIRFLOW-3562> is trying to
> > > remove
> > > > > the dependency of dagbag from webserver, which is awesome--we
> wanted
> > > > badly
> > > > > but never got to start work on. However when I looked at some
> > subtasks
> > > of
> > > > > it, which try to remove dagbag dependency from each endpoint, I
> found
> > > the
> > > > > way we remove the dependency of dagbag is not very ideal. For
> example
> > > > this
> > > > > PR <https://github.com/apache/airflow/pull/4867/files> will
> require
> > us
> > > > to
> > > > > parse the dag file each time we hit the endpoint.
> > > > >
> > > > > The counter argument: this PR removes the need for the confusing
> > > > "Refresh"
> > > > > button from the UI, and in general you only pay the cost for the
> > > > expensive
> > > > > DAGs when you ask about them. (I don't know what/when we call the
> > > > > /pickle_info endpoint of the top of my head)
> > > > >
> > > > > This end point may be one to hold off on (as it can ask for
> multiple
> > > > dags)
> > > > > but there are some that def don't need a full dag bag or to even
> > parse
> > > > the
> > > > > dag file, the current DAG model has enough info.
> > > > >
> > > > > >
> > > > > >
> > > > > > If we go down this path, we indeed can get rid of the dagbag
> > > dependency
> > > > > easily, but we will have to 1. increase the DB load( not too
> > concerning
> > > > at
> > > > > the moment ), 2. wait the DAG file to be parsed before getting the
> > page
> > > > > back, potentially multiple times. DAG file can sometimes take
> quite a
> > > > while
> > > > > to parse, e.g. we have some framework DAG files generating large
> > number
> > > > of
> > > > > DAGs from some static config files or even jupyter notebooks and
> they
> > > can
> > > > > take 30+ seconds to parse. Yes we don't like large DAG files but
> > people
> > > > do
> > > > > see the beauty of code as config and sometimes heavily
> abuseleverage
> > > it.
> > > > > Assuming all users have the same nice small python file that can be
> > > > parsed
> > > > > fast, I'm still a bit worried about this approach. Continuing on
> this
> > > > path
> > > > > means we've chosen DagModel to be the serialized representation of
> > DAG
> > > > and
> > > > > DB columns to hold different properties--it can be one candidate
> but
> > I
> > > > > don't know if we should settle on that now. I would personally
> > prefer a
> > > > > more compact, e.g. JSON5, and easy to scale representation( such
> that
> > > > > serializing new fields != DB upgrade).
> > > > >
> > > > > Do you mean https://json5.org/ or is this a typo? That might be
> okay
> > > for
> > > > > a nicer user front end, but the "canonical" version stored in the
> DB
> > > > should
> > > > > be something "plainer" like just JSON.
> > > > >
> > > > > I'm not sure that "serializing new fields != DB upgrade" is that
> big
> > > of a
> > > > > concern, as we don't add fields that often. One possible way of
> > dealing
> > > > > with it if we do is to have a hybrid approach - a few distinct
> > columns,
> > > > but
> > > > > then a JSON blob. (and if we were only to support postgres we could
> > > just
> > > > > use JSONb. But I think our friends at Google may object ;) )
> > > > >
> > > > > Adding a new column in a DB migration with a default NULL shouldn't
> > be
> > > an
> > > > > expensive operation, or difficult to achieve.
> > > > >
> > > > >
> > > > > >
> > > > > > In my imagination we would have to collect the list of dynamic
> > > features
> > > > > depending on unserializable fields of a DAG and start a
> > discussion/vote
> > > > on
> > > > > dropping support of them( I'm working on this but if anyone has
> > already
> > > > > done so please take over), decide on the serialized representation
> > of a
> > > > DAG
> > > > > and then replace dagbag with it in webserver. Per previous
> discussion
> > > and
> > > > > some offline discussions with Dan, one future of DAG serialization
> > > that I
> > > > > like would look similar to this:
> > > > > >
> > > > >
> > > > > > https://imgur.com/ncqqQgc
> > > > >
> > > > > Something I've thought about before for other things was to embed
> an
> > > API
> > > > > server _into_ the scheduler - this would be useful for k8s
> > > healthchecks,
> > > > > native Prometheus metrics without needed statsd bridge, and could
> > have
> > > > > endpoints to get information such as this directly.
> > > > >
> > > > > I was thinking it would be _in_ the scheduler process using either
> > > > threads
> > > > > (ick. Python's still got a GIL doesn't it?) or using async/twisted
> > etc.
> > > > > (not a side-car process like we have with the logs webserver for
> > > `airflow
> > > > > worker`).
> > > > >
> > > > > (This is possibly an unrelated discussion, but might be worth
> talking
> > > > > about?)
> > > > >
> > > > > > We can still discuss/vote which approach we want to take but I
> > don't
> > > > > want the door to above design to be shut right now or we have to
> > spend
> > > a
> > > > > lot effort switch path later.
> > > > > >
> > > > > > Bas and Peter, I'm very sorry to extend the discussion but I do
> > think
> > > > > this is tightly related to the AIP and PRs behind it. And my
> sincere
> > > > > apology for bringing this up so late( I only pull the open PR list
> > > > > occasionally, if there's a way to subscribe to new PR event I'd
> love
> > to
> > > > > know how).
> > > > >
> > > > > It's noisy, but you can subscribe to commits@airflow.apache.org
> (but
> > > be
> > > > > warned, this also includes all Jira tickets, edits of every comment
> > on
> > > > > github etc.).
> > > > >
> > > > >
> > > > > >
> > > > > > Cheers,
> > > > > > Kevin Y
> > > > > >
> > > > > > On Thu, Feb 28, 2019 at 1:36 PM Peter van t Hof <
> > > pjrvanthof@gmail.com
> > > > > <ma...@gmail.com>> wrote:
> > > > > > Hi all,
> > > > > >
> > > > > > Just some comments one the point Bolke dit give in relation of my
> > PR.
> > > > > >
> > > > > > At first, the main focus is: making the webserver stateless.
> > > > > >
> > > > > > > 1) Make the webserver stateless: needs the graph of the
> *current*
> > > dag
> > > > > >
> > > > > > This is the main goal but for this a lot more PR’s will be coming
> > > once
> > > > > my current is merged. For edges and graph view this is covered in
> my
> > PR
> > > > > already.
> > > > > >
> > > > > > > 2) Version dags: for consistency mainly and not requiring
> parsing
> > > of
> > > > > the
> > > > > > > dag on every loop
> > > > > >
> > > > > > In my PR the historical graphs will be stored for each DagRun.
> This
> > > > > means that you can see if an older DagRun was the same graph
> > structure,
> > > > > even if some tasks does not exists anymore in the current graph.
> > > > Especially
> > > > > for dynamic DAG’s this is very useful.
> > > > > >
> > > > > > > 3) Make the scheduler not require DAG files. This could be done
> > if
> > > > the
> > > > > > > edges contain all information when to trigger the next task. We
> > can
> > > > > then
> > > > > > > have event driven dag parsing outside of the scheduler loop,
> ie.
> > by
> > > > the
> > > > > > > cli. Storage can also be somewhere else (git, artifactory,
> > > > filesystem,
> > > > > > > whatever).
> > > > > >
> > > > > > The scheduler is almost untouched in this PR. The only thing that
> > is
> > > > > added is that this edges are saved to the database but the
> scheduling
> > > > > itself din’t change. The scheduler depends now still on the DAG
> > object.
> > > > > >
> > > > > > > 4) Fully serialise the dag so it becomes transferable to
> workers
> > > > > >
> > > > > > It nice to see that people has a lot of idea’s about this. But as
> > > Fokko
> > > > > already mentioned this is out of scope for the issue what we are
> > trying
> > > > to
> > > > > solve. I also have some idea’s about this but I like to limit this
> > > PR/AIP
> > > > > to the webserver.
> > > > > >
> > > > > > For now my PR does solve 1 and 2 and the rest of the behaviour
> > (like
> > > > > scheduling) is untouched.
> > > > > >
> > > > > > Gr,
> > > > > > Peter
> > > > > >
> > > > >
> > > > >
> > > >
> > >
> >
>

Re: [DISCUSS] AIP-12 Persist DAG into DB

["Driesprong, Fokko" <fo...@driesprong.frl>]

Thanks Kevin for opening the discussion. I think it is important to have a
clear overview on how to approach the AIP.

First of all, how many DAGs do we have that take 30s to parse? I consider
this bad practice, and this would also result in an unworkable situation
with the current setup of Airflow since it will take a lot of resources on
the webserver/scheduler, and the whole system will become unresponsive. I
will be hard to cope with such DAGs in general.

The idea from the AIP is to have the versioned version of the dag in the
DB, so in the end, you won't need to parse the whole thing every time. Only
when you trigger a DAG, or when you want to see the current status of the
dag.

Like stated earlier, I strongly feel we shouldn't serialize the DAGs as
JSON(5) or pickles in general. For me, this is deferring the pain of
setting up a structure of the DAG object itself.
Having the DAG denormalized in the database will give us cleaner storage of
our DAG. We can, for example, enforce fields by making them not null, so we
know that is something is off at write time, instead of read. Furthermore,
we're missing logical types such as dates, which we efficiently can query
using the indices of the database.
Also, with all serialization formats, evolution isn't trivial. Consider the
situations when:
- We're introducing a new field, and it might be null, therefore we need to
bake in all kinds of logic into the Airflow code, which you don't want.
With proper migration scripts, you could prefill these fields, and make
them not null.
- Changing the models, for example, you still can't change a string-type
into a integer with adding custom logic. In this case, the reviewer needs
to be extra careful that there are no breaking changes introduced. Right
now we're doing minimal forward- and backward compatibilitytesting.

In the case we get too many migrations, we could also squash (some of them)
when preparing the release.

Personally, I don't think the serialization is the issue here. As Max
already mentioned, it is the optimal balance of (de)normalization. From the
user perspective, the serialization won't change much of the behaviour of
Airflow.

For me, instead of having `DAG.serialize()` and `DAG.deser(version)` is not
the ideal approach. But it might be that we're on the same page :-) I
believe it should be something like `DagRun.find('fokkos_dag',
datetime(2018, 03, 01))` and construct the correct version of the dag.
Since there is an uniqueness constrain on dag_id, datetime, this will
always return the same dag. You will get the versioned DagRun as it ran
that time. Serializing the fields adn storing them in the database should
happen transparently when you update the DAG object. When you run a dag,
you'll parse the dag, and then run it. `Dag().create_dagrun(...)`, this
will create a DagRun as the name suggests, if the version of the dag still
exists in the database, it will reuse that one, otherwise it will create a
new version of the DAG (with all the operators etc). In this sense the
version of the DAGs should be done within the Dag(Run).

The versioning will change the behavour from a user perspective. Right now
we store only a single version. For example, the poor mans backfilling
won't work anymore. This is clearing the state from past&future, up- and
downstream, and let it catch up again.
In this case, the old version of the DAG won't exists anymore, and
potentially there are tasks that aren't in the code anymore. In this case
we need to clear the version of the dag, and rerun it with the latest
version `DagRun.find('fokkos_dag', datetime(2018, 03, 01)).clear()`. How we
are going to do clear's downstram in the middle of the dag, that is
something I still have to figure out. Because potentially there are tasks
that can't be rerun because the underlying Python code has changed, both on
user level as on Airflow level. It will be impossible to get these features
pure in that sense.
I would not suggest adding a new status in here, indicating that the task
can't be rerun since it isn't part of the DAG anymore. We have to find the
balance here in adding complexity (also to the scheduler) and features that
we need to introduce to help the user.

Cheers, Fokko

Ps. Jarek, interesting idea. It shouldn't be too hard to make Airflow more
k8s native. You could package your dags within your container, and do a
rolling update. Add the DAGs as the last layer, and then point the DAGs
folder to the same location. The hard part here is that you need to
gracefuly restart the workers. Currently AFAIK the signals given to the pod
aren't respected. So when the scheduler/webserver/worker receives a
SIGTERM, it should stop the jobs nicely and then exit the container, before
k8s kills the container using a SIGKILL.  This will be challenging with the
workers, which they are potentially long-running. Maybe stop kicking off
new jobs, and let the old ones finish, will be good enough, but then we
need to increase the standard kill timeout substantially. Having this would
also enable the autoscaling of the workers.



Op za 9 mrt. 2019 om 19:07 schreef Maxime Beauchemin <
maximebeauchemin@gmail.com>:

> I want to raise the question of the amount of normalization we want to use
> here as it seems the to be an area that needs more attention.
>
> The SIP suggest having DAG blobs, task blobs and edges (call it the
> fairly-normalized). I also like the idea of all-encompassing (call it
> very-denormalized) DAG blobs as it seems easier to manage in terms of
> versioning. The question here is whether we go with one of these method
> exclusively, something in-between or even a hybrid approach (redundant
> blobs that use different level of normalization).
>
> It's nice and simple to just push or pull DAG atomic objects with a version
> stamp on it. It's clearly simpler than dealing with 3 versioned tables
> (dag, tasks, edges). There are a lot of pros/cons, and they become more
> apparent with the perspective of very large DAGs. If the web server is
> building a "task details page", using the "fairly-normalized" model, it can
> just pull what it needs instead of pulling the large DAG blob. Similarly,
> if building a sub-tree view (a subset of the DAG), perhaps it can only
> retrieve what it needs. But if you need the whole DAG (say for the
> scheduler use case) then you're dealing with more complex SQL/ORM
> operations (joins hopefully, or multiple db round trips)
>
> Now maybe the right approach is more something like 2 tables: DAG and
> task_details, where edges keys are denormalized into DAG (arguably that's a
> few KBs at most, even for large DAGs), and maybe the DAG object has most of
> the high level task metadata information (operator, name, baseoperator key
> attrs), and task_details has the big blobs (SQL code). This is probably a
> nice compromise, the question becomes "how much task-level detail do we
> store in the DAG-centric blog?", probably not much to keep the DAG objects
> as small as possible. The main downside here is that you cannot have the
> database join and have to do 2 round trips to reconstruct a DAG object
> (fetch the DAG, parse the object to get the list of tasks, and then run
> another db query to get those task details).
>
> To resume, I'd qualify the more normalized approach as the most proper, but
> also the more complex. It'll shine in specific cases around large DAGs. If
> we have the proper abstractions (methods like DAG.serialize(),
> DAG.deser(version)) then I guess that's not an issue.
>
> Max
>
> On Fri, Mar 8, 2019 at 5:21 PM Kevin Yang <yr...@gmail.com> wrote:
>
> > Hi Julian, I'm definitely aligned with you guys on making the webserver
> > independent of DAG parsing, just the end goal to me would be to build a
> > complete story around serializing DAG--and move with the story in mind. I
> > feel like you guys may already have a list of dynamic features we need to
> > deprecate/change, if that is the case feel free to open the discussion on
> > what we do to them with DAG serialization.
> >
> > Julian, Ash, Dan, on 2nd thought I do agree that if we can meet the
> > requirements Dan mentioned, it would be nice to have them stored in the
> DB.
> > Some combined solutions like having a column of serialized graph in the
> > serialized dag table can potentially meet all requirements. What format
> we
> > end up using to represent DAG between components is now less important
> > IMO--fine to refactor those endpoints only need DagModel to use only
> > DagModel, easy to do a batch replacement if we decide otherwise later.
> More
> > important is to define this source of truth for serialized DAG.
> >
> > Ash, ty for the email list, I'll tune my filters accordingly :D I'm
> leaning
> > towards having a separate process for the parser so we got no scheduler
> > dependency etc for this parser but we can discuss this in another thread.
> >
> > On Fri, Mar 8, 2019 at 8:57 AM Dan Davydov <ddavydov@twitter.com.invalid
> >
> > wrote:
> >
> > > >
> > > > Personally I don’t understand why people are pushing for a JSON-based
> > DAG
> > > > representation
> > >
> > > It sounds like you agree that DAGs should be serialized (just in the DB
> > > instead of JSON), so will only address why JSON is better than MySQL
> (AKA
> > > serializing at the DAG level vs the task level) as far as I can see,
> and
> > > not why we need serialization. If you zoom out and look at all the use
> > > cases of serialized DAGs, e.g. having the scheduler use them instead of
> > > parsing DAGs directly, then it becomes clear that we need all
> appropriate
> > > metadata in these DAGs, (operator params, DAG properties, etc), in
> which
> > > case it's not clear how it will fit nicely into a DB table (unless you
> > > wanted to do something like (parent_task_id, task_id, task_params),
> also
> > > keep in mind that we will need to store different versions of each DAG
> in
> > > the future so that we can ensure consistency in a dagrun, i.e. each
> task
> > in
> > > a dagrun uses the same version of a DAG.
> > >
> > > I think some of our requirements should be:
> > > 1. The data model will lead to acceptable performance in all of its
> > > consumers (scheduler, webserver, workers), i.e. no n+1 access patterns
> > (my
> > > biggest concern about serializing at task level as you propose vs at
> DAG
> > > level)
> > > 2. We can have versioning of serialized DAGs
> > > 3. The ability to separate DAGs into their own data store (e.g. no
> > reliance
> > > on joins between the new table and the old one)
> > > 4. One source of truth/serialized representation for DAGs (currently we
> > > have SimpleDAG)
> > >
> > > If we can full-fill all of these requirements and serialize at the task
> > > level rather than the DAG level in the DB, then I agree that probably
> > makes
> > > more sense.
> > >
> > >
> > > > In the proposed PR’s we (Peter, Bas and me) aim to avoid re-parsing
> DAG
> > > > files by querying all the required information from the database. In
> > one
> > > or
> > > > two cases this may however not be possible, in which case we might
> > either
> > > > have to fall back on the DAG file or add the missing information into
> > the
> > > > database. We can tackle these problems as we encounter them.
> > >
> > > I think you would have the support of many of committers in removing
> any
> > > use-cases that stand in the way of full serialization, that being said
> if
> > > we need to remove features we need to do this carefully and
> thoughtfully,
> > > and ideally with proposed alternatives/work-arounds to cover the
> > removals.
> > >
> > > The counter argument: this PR removes the need for the confusing
> > "Refresh"
> > > > button from the UI, and in general you only pay the cost for the
> > > expensive
> > > > DAGs when you ask about them. (I don't know what/when we call the
> > > > /pickle_info endpoint of the top of my head)
> > >
> > > Probably worth splitting out into a separate thread, but I'm actually
> not
> > > sure the refresh button does anything, I think we should double
> check...
> > I
> > > think about 2 years ago there was a commit made that made gunicorn
> > > webservers automatically rotate underneath flask (each one would
> reparse
> > > the DAGbag). Even if it works we should probably remove it since the
> > > webserver refresh interval is pretty fast, and it just causes confusion
> > to
> > > users and implies that the DAGs are not refreshed automatically.
> > >
> > > Do you mean https://json5.org/ or is this a typo? That might be okay
> > for a
> > > > nicer user front end, but the "canonical" version stored in the DB
> > should
> > > > be something "plainer" like just JSON.
> > >
> > > I think he got this from my reply, and it was just an example, but you
> > are
> > > right, I agree JSON would be better than JSON5.
> > >
> > > On Fri, Mar 8, 2019 at 8:53 AM Ash Berlin-Taylor <as...@apache.org>
> wrote:
> > >
> > > > Comments inline.
> > > >
> > > > > On 8 Mar 2019, at 11:28, Kevin Yang <yr...@gmail.com> wrote:
> > > > >
> > > > > Hi all,
> > > > > When I was preparing some work related to this AIP I found
> something
> > > > very concerning. I noticed this JIRA ticket <
> > > > https://issues.apache.org/jira/browse/AIRFLOW-3562> is trying to
> > remove
> > > > the dependency of dagbag from webserver, which is awesome--we wanted
> > > badly
> > > > but never got to start work on. However when I looked at some
> subtasks
> > of
> > > > it, which try to remove dagbag dependency from each endpoint, I found
> > the
> > > > way we remove the dependency of dagbag is not very ideal. For example
> > > this
> > > > PR <https://github.com/apache/airflow/pull/4867/files> will require
> us
> > > to
> > > > parse the dag file each time we hit the endpoint.
> > > >
> > > > The counter argument: this PR removes the need for the confusing
> > > "Refresh"
> > > > button from the UI, and in general you only pay the cost for the
> > > expensive
> > > > DAGs when you ask about them. (I don't know what/when we call the
> > > > /pickle_info endpoint of the top of my head)
> > > >
> > > > This end point may be one to hold off on (as it can ask for multiple
> > > dags)
> > > > but there are some that def don't need a full dag bag or to even
> parse
> > > the
> > > > dag file, the current DAG model has enough info.
> > > >
> > > > >
> > > > >
> > > > > If we go down this path, we indeed can get rid of the dagbag
> > dependency
> > > > easily, but we will have to 1. increase the DB load( not too
> concerning
> > > at
> > > > the moment ), 2. wait the DAG file to be parsed before getting the
> page
> > > > back, potentially multiple times. DAG file can sometimes take quite a
> > > while
> > > > to parse, e.g. we have some framework DAG files generating large
> number
> > > of
> > > > DAGs from some static config files or even jupyter notebooks and they
> > can
> > > > take 30+ seconds to parse. Yes we don't like large DAG files but
> people
> > > do
> > > > see the beauty of code as config and sometimes heavily abuseleverage
> > it.
> > > > Assuming all users have the same nice small python file that can be
> > > parsed
> > > > fast, I'm still a bit worried about this approach. Continuing on this
> > > path
> > > > means we've chosen DagModel to be the serialized representation of
> DAG
> > > and
> > > > DB columns to hold different properties--it can be one candidate but
> I
> > > > don't know if we should settle on that now. I would personally
> prefer a
> > > > more compact, e.g. JSON5, and easy to scale representation( such that
> > > > serializing new fields != DB upgrade).
> > > >
> > > > Do you mean https://json5.org/ or is this a typo? That might be okay
> > for
> > > > a nicer user front end, but the "canonical" version stored in the DB
> > > should
> > > > be something "plainer" like just JSON.
> > > >
> > > > I'm not sure that "serializing new fields != DB upgrade" is that big
> > of a
> > > > concern, as we don't add fields that often. One possible way of
> dealing
> > > > with it if we do is to have a hybrid approach - a few distinct
> columns,
> > > but
> > > > then a JSON blob. (and if we were only to support postgres we could
> > just
> > > > use JSONb. But I think our friends at Google may object ;) )
> > > >
> > > > Adding a new column in a DB migration with a default NULL shouldn't
> be
> > an
> > > > expensive operation, or difficult to achieve.
> > > >
> > > >
> > > > >
> > > > > In my imagination we would have to collect the list of dynamic
> > features
> > > > depending on unserializable fields of a DAG and start a
> discussion/vote
> > > on
> > > > dropping support of them( I'm working on this but if anyone has
> already
> > > > done so please take over), decide on the serialized representation
> of a
> > > DAG
> > > > and then replace dagbag with it in webserver. Per previous discussion
> > and
> > > > some offline discussions with Dan, one future of DAG serialization
> > that I
> > > > like would look similar to this:
> > > > >
> > > >
> > > > > https://imgur.com/ncqqQgc
> > > >
> > > > Something I've thought about before for other things was to embed an
> > API
> > > > server _into_ the scheduler - this would be useful for k8s
> > healthchecks,
> > > > native Prometheus metrics without needed statsd bridge, and could
> have
> > > > endpoints to get information such as this directly.
> > > >
> > > > I was thinking it would be _in_ the scheduler process using either
> > > threads
> > > > (ick. Python's still got a GIL doesn't it?) or using async/twisted
> etc.
> > > > (not a side-car process like we have with the logs webserver for
> > `airflow
> > > > worker`).
> > > >
> > > > (This is possibly an unrelated discussion, but might be worth talking
> > > > about?)
> > > >
> > > > > We can still discuss/vote which approach we want to take but I
> don't
> > > > want the door to above design to be shut right now or we have to
> spend
> > a
> > > > lot effort switch path later.
> > > > >
> > > > > Bas and Peter, I'm very sorry to extend the discussion but I do
> think
> > > > this is tightly related to the AIP and PRs behind it. And my sincere
> > > > apology for bringing this up so late( I only pull the open PR list
> > > > occasionally, if there's a way to subscribe to new PR event I'd love
> to
> > > > know how).
> > > >
> > > > It's noisy, but you can subscribe to commits@airflow.apache.org (but
> > be
> > > > warned, this also includes all Jira tickets, edits of every comment
> on
> > > > github etc.).
> > > >
> > > >
> > > > >
> > > > > Cheers,
> > > > > Kevin Y
> > > > >
> > > > > On Thu, Feb 28, 2019 at 1:36 PM Peter van t Hof <
> > pjrvanthof@gmail.com
> > > > <ma...@gmail.com>> wrote:
> > > > > Hi all,
> > > > >
> > > > > Just some comments one the point Bolke dit give in relation of my
> PR.
> > > > >
> > > > > At first, the main focus is: making the webserver stateless.
> > > > >
> > > > > > 1) Make the webserver stateless: needs the graph of the *current*
> > dag
> > > > >
> > > > > This is the main goal but for this a lot more PR’s will be coming
> > once
> > > > my current is merged. For edges and graph view this is covered in my
> PR
> > > > already.
> > > > >
> > > > > > 2) Version dags: for consistency mainly and not requiring parsing
> > of
> > > > the
> > > > > > dag on every loop
> > > > >
> > > > > In my PR the historical graphs will be stored for each DagRun. This
> > > > means that you can see if an older DagRun was the same graph
> structure,
> > > > even if some tasks does not exists anymore in the current graph.
> > > Especially
> > > > for dynamic DAG’s this is very useful.
> > > > >
> > > > > > 3) Make the scheduler not require DAG files. This could be done
> if
> > > the
> > > > > > edges contain all information when to trigger the next task. We
> can
> > > > then
> > > > > > have event driven dag parsing outside of the scheduler loop, ie.
> by
> > > the
> > > > > > cli. Storage can also be somewhere else (git, artifactory,
> > > filesystem,
> > > > > > whatever).
> > > > >
> > > > > The scheduler is almost untouched in this PR. The only thing that
> is
> > > > added is that this edges are saved to the database but the scheduling
> > > > itself din’t change. The scheduler depends now still on the DAG
> object.
> > > > >
> > > > > > 4) Fully serialise the dag so it becomes transferable to workers
> > > > >
> > > > > It nice to see that people has a lot of idea’s about this. But as
> > Fokko
> > > > already mentioned this is out of scope for the issue what we are
> trying
> > > to
> > > > solve. I also have some idea’s about this but I like to limit this
> > PR/AIP
> > > > to the webserver.
> > > > >
> > > > > For now my PR does solve 1 and 2 and the rest of the behaviour
> (like
> > > > scheduling) is untouched.
> > > > >
> > > > > Gr,
> > > > > Peter
> > > > >
> > > >
> > > >
> > >
> >
>

Re: [DISCUSS] AIP-12 Persist DAG into DB

[Maxime Beauchemin <ma...@gmail.com>]

I want to raise the question of the amount of normalization we want to use
here as it seems the to be an area that needs more attention.

The SIP suggest having DAG blobs, task blobs and edges (call it the
fairly-normalized). I also like the idea of all-encompassing (call it
very-denormalized) DAG blobs as it seems easier to manage in terms of
versioning. The question here is whether we go with one of these method
exclusively, something in-between or even a hybrid approach (redundant
blobs that use different level of normalization).

It's nice and simple to just push or pull DAG atomic objects with a version
stamp on it. It's clearly simpler than dealing with 3 versioned tables
(dag, tasks, edges). There are a lot of pros/cons, and they become more
apparent with the perspective of very large DAGs. If the web server is
building a "task details page", using the "fairly-normalized" model, it can
just pull what it needs instead of pulling the large DAG blob. Similarly,
if building a sub-tree view (a subset of the DAG), perhaps it can only
retrieve what it needs. But if you need the whole DAG (say for the
scheduler use case) then you're dealing with more complex SQL/ORM
operations (joins hopefully, or multiple db round trips)

Now maybe the right approach is more something like 2 tables: DAG and
task_details, where edges keys are denormalized into DAG (arguably that's a
few KBs at most, even for large DAGs), and maybe the DAG object has most of
the high level task metadata information (operator, name, baseoperator key
attrs), and task_details has the big blobs (SQL code). This is probably a
nice compromise, the question becomes "how much task-level detail do we
store in the DAG-centric blog?", probably not much to keep the DAG objects
as small as possible. The main downside here is that you cannot have the
database join and have to do 2 round trips to reconstruct a DAG object
(fetch the DAG, parse the object to get the list of tasks, and then run
another db query to get those task details).

To resume, I'd qualify the more normalized approach as the most proper, but
also the more complex. It'll shine in specific cases around large DAGs. If
we have the proper abstractions (methods like DAG.serialize(),
DAG.deser(version)) then I guess that's not an issue.

Max

On Fri, Mar 8, 2019 at 5:21 PM Kevin Yang <yr...@gmail.com> wrote:

> Hi Julian, I'm definitely aligned with you guys on making the webserver
> independent of DAG parsing, just the end goal to me would be to build a
> complete story around serializing DAG--and move with the story in mind. I
> feel like you guys may already have a list of dynamic features we need to
> deprecate/change, if that is the case feel free to open the discussion on
> what we do to them with DAG serialization.
>
> Julian, Ash, Dan, on 2nd thought I do agree that if we can meet the
> requirements Dan mentioned, it would be nice to have them stored in the DB.
> Some combined solutions like having a column of serialized graph in the
> serialized dag table can potentially meet all requirements. What format we
> end up using to represent DAG between components is now less important
> IMO--fine to refactor those endpoints only need DagModel to use only
> DagModel, easy to do a batch replacement if we decide otherwise later. More
> important is to define this source of truth for serialized DAG.
>
> Ash, ty for the email list, I'll tune my filters accordingly :D I'm leaning
> towards having a separate process for the parser so we got no scheduler
> dependency etc for this parser but we can discuss this in another thread.
>
> On Fri, Mar 8, 2019 at 8:57 AM Dan Davydov <dd...@twitter.com.invalid>
> wrote:
>
> > >
> > > Personally I don’t understand why people are pushing for a JSON-based
> DAG
> > > representation
> >
> > It sounds like you agree that DAGs should be serialized (just in the DB
> > instead of JSON), so will only address why JSON is better than MySQL (AKA
> > serializing at the DAG level vs the task level) as far as I can see, and
> > not why we need serialization. If you zoom out and look at all the use
> > cases of serialized DAGs, e.g. having the scheduler use them instead of
> > parsing DAGs directly, then it becomes clear that we need all appropriate
> > metadata in these DAGs, (operator params, DAG properties, etc), in which
> > case it's not clear how it will fit nicely into a DB table (unless you
> > wanted to do something like (parent_task_id, task_id, task_params), also
> > keep in mind that we will need to store different versions of each DAG in
> > the future so that we can ensure consistency in a dagrun, i.e. each task
> in
> > a dagrun uses the same version of a DAG.
> >
> > I think some of our requirements should be:
> > 1. The data model will lead to acceptable performance in all of its
> > consumers (scheduler, webserver, workers), i.e. no n+1 access patterns
> (my
> > biggest concern about serializing at task level as you propose vs at DAG
> > level)
> > 2. We can have versioning of serialized DAGs
> > 3. The ability to separate DAGs into their own data store (e.g. no
> reliance
> > on joins between the new table and the old one)
> > 4. One source of truth/serialized representation for DAGs (currently we
> > have SimpleDAG)
> >
> > If we can full-fill all of these requirements and serialize at the task
> > level rather than the DAG level in the DB, then I agree that probably
> makes
> > more sense.
> >
> >
> > > In the proposed PR’s we (Peter, Bas and me) aim to avoid re-parsing DAG
> > > files by querying all the required information from the database. In
> one
> > or
> > > two cases this may however not be possible, in which case we might
> either
> > > have to fall back on the DAG file or add the missing information into
> the
> > > database. We can tackle these problems as we encounter them.
> >
> > I think you would have the support of many of committers in removing any
> > use-cases that stand in the way of full serialization, that being said if
> > we need to remove features we need to do this carefully and thoughtfully,
> > and ideally with proposed alternatives/work-arounds to cover the
> removals.
> >
> > The counter argument: this PR removes the need for the confusing
> "Refresh"
> > > button from the UI, and in general you only pay the cost for the
> > expensive
> > > DAGs when you ask about them. (I don't know what/when we call the
> > > /pickle_info endpoint of the top of my head)
> >
> > Probably worth splitting out into a separate thread, but I'm actually not
> > sure the refresh button does anything, I think we should double check...
> I
> > think about 2 years ago there was a commit made that made gunicorn
> > webservers automatically rotate underneath flask (each one would reparse
> > the DAGbag). Even if it works we should probably remove it since the
> > webserver refresh interval is pretty fast, and it just causes confusion
> to
> > users and implies that the DAGs are not refreshed automatically.
> >
> > Do you mean https://json5.org/ or is this a typo? That might be okay
> for a
> > > nicer user front end, but the "canonical" version stored in the DB
> should
> > > be something "plainer" like just JSON.
> >
> > I think he got this from my reply, and it was just an example, but you
> are
> > right, I agree JSON would be better than JSON5.
> >
> > On Fri, Mar 8, 2019 at 8:53 AM Ash Berlin-Taylor <as...@apache.org> wrote:
> >
> > > Comments inline.
> > >
> > > > On 8 Mar 2019, at 11:28, Kevin Yang <yr...@gmail.com> wrote:
> > > >
> > > > Hi all,
> > > > When I was preparing some work related to this AIP I found something
> > > very concerning. I noticed this JIRA ticket <
> > > https://issues.apache.org/jira/browse/AIRFLOW-3562> is trying to
> remove
> > > the dependency of dagbag from webserver, which is awesome--we wanted
> > badly
> > > but never got to start work on. However when I looked at some subtasks
> of
> > > it, which try to remove dagbag dependency from each endpoint, I found
> the
> > > way we remove the dependency of dagbag is not very ideal. For example
> > this
> > > PR <https://github.com/apache/airflow/pull/4867/files> will require us
> > to
> > > parse the dag file each time we hit the endpoint.
> > >
> > > The counter argument: this PR removes the need for the confusing
> > "Refresh"
> > > button from the UI, and in general you only pay the cost for the
> > expensive
> > > DAGs when you ask about them. (I don't know what/when we call the
> > > /pickle_info endpoint of the top of my head)
> > >
> > > This end point may be one to hold off on (as it can ask for multiple
> > dags)
> > > but there are some that def don't need a full dag bag or to even parse
> > the
> > > dag file, the current DAG model has enough info.
> > >
> > > >
> > > >
> > > > If we go down this path, we indeed can get rid of the dagbag
> dependency
> > > easily, but we will have to 1. increase the DB load( not too concerning
> > at
> > > the moment ), 2. wait the DAG file to be parsed before getting the page
> > > back, potentially multiple times. DAG file can sometimes take quite a
> > while
> > > to parse, e.g. we have some framework DAG files generating large number
> > of
> > > DAGs from some static config files or even jupyter notebooks and they
> can
> > > take 30+ seconds to parse. Yes we don't like large DAG files but people
> > do
> > > see the beauty of code as config and sometimes heavily abuseleverage
> it.
> > > Assuming all users have the same nice small python file that can be
> > parsed
> > > fast, I'm still a bit worried about this approach. Continuing on this
> > path
> > > means we've chosen DagModel to be the serialized representation of DAG
> > and
> > > DB columns to hold different properties--it can be one candidate but I
> > > don't know if we should settle on that now. I would personally prefer a
> > > more compact, e.g. JSON5, and easy to scale representation( such that
> > > serializing new fields != DB upgrade).
> > >
> > > Do you mean https://json5.org/ or is this a typo? That might be okay
> for
> > > a nicer user front end, but the "canonical" version stored in the DB
> > should
> > > be something "plainer" like just JSON.
> > >
> > > I'm not sure that "serializing new fields != DB upgrade" is that big
> of a
> > > concern, as we don't add fields that often. One possible way of dealing
> > > with it if we do is to have a hybrid approach - a few distinct columns,
> > but
> > > then a JSON blob. (and if we were only to support postgres we could
> just
> > > use JSONb. But I think our friends at Google may object ;) )
> > >
> > > Adding a new column in a DB migration with a default NULL shouldn't be
> an
> > > expensive operation, or difficult to achieve.
> > >
> > >
> > > >
> > > > In my imagination we would have to collect the list of dynamic
> features
> > > depending on unserializable fields of a DAG and start a discussion/vote
> > on
> > > dropping support of them( I'm working on this but if anyone has already
> > > done so please take over), decide on the serialized representation of a
> > DAG
> > > and then replace dagbag with it in webserver. Per previous discussion
> and
> > > some offline discussions with Dan, one future of DAG serialization
> that I
> > > like would look similar to this:
> > > >
> > >
> > > > https://imgur.com/ncqqQgc
> > >
> > > Something I've thought about before for other things was to embed an
> API
> > > server _into_ the scheduler - this would be useful for k8s
> healthchecks,
> > > native Prometheus metrics without needed statsd bridge, and could have
> > > endpoints to get information such as this directly.
> > >
> > > I was thinking it would be _in_ the scheduler process using either
> > threads
> > > (ick. Python's still got a GIL doesn't it?) or using async/twisted etc.
> > > (not a side-car process like we have with the logs webserver for
> `airflow
> > > worker`).
> > >
> > > (This is possibly an unrelated discussion, but might be worth talking
> > > about?)
> > >
> > > > We can still discuss/vote which approach we want to take but I don't
> > > want the door to above design to be shut right now or we have to spend
> a
> > > lot effort switch path later.
> > > >
> > > > Bas and Peter, I'm very sorry to extend the discussion but I do think
> > > this is tightly related to the AIP and PRs behind it. And my sincere
> > > apology for bringing this up so late( I only pull the open PR list
> > > occasionally, if there's a way to subscribe to new PR event I'd love to
> > > know how).
> > >
> > > It's noisy, but you can subscribe to commits@airflow.apache.org (but
> be
> > > warned, this also includes all Jira tickets, edits of every comment on
> > > github etc.).
> > >
> > >
> > > >
> > > > Cheers,
> > > > Kevin Y
> > > >
> > > > On Thu, Feb 28, 2019 at 1:36 PM Peter van t Hof <
> pjrvanthof@gmail.com
> > > <ma...@gmail.com>> wrote:
> > > > Hi all,
> > > >
> > > > Just some comments one the point Bolke dit give in relation of my PR.
> > > >
> > > > At first, the main focus is: making the webserver stateless.
> > > >
> > > > > 1) Make the webserver stateless: needs the graph of the *current*
> dag
> > > >
> > > > This is the main goal but for this a lot more PR’s will be coming
> once
> > > my current is merged. For edges and graph view this is covered in my PR
> > > already.
> > > >
> > > > > 2) Version dags: for consistency mainly and not requiring parsing
> of
> > > the
> > > > > dag on every loop
> > > >
> > > > In my PR the historical graphs will be stored for each DagRun. This
> > > means that you can see if an older DagRun was the same graph structure,
> > > even if some tasks does not exists anymore in the current graph.
> > Especially
> > > for dynamic DAG’s this is very useful.
> > > >
> > > > > 3) Make the scheduler not require DAG files. This could be done if
> > the
> > > > > edges contain all information when to trigger the next task. We can
> > > then
> > > > > have event driven dag parsing outside of the scheduler loop, ie. by
> > the
> > > > > cli. Storage can also be somewhere else (git, artifactory,
> > filesystem,
> > > > > whatever).
> > > >
> > > > The scheduler is almost untouched in this PR. The only thing that is
> > > added is that this edges are saved to the database but the scheduling
> > > itself din’t change. The scheduler depends now still on the DAG object.
> > > >
> > > > > 4) Fully serialise the dag so it becomes transferable to workers
> > > >
> > > > It nice to see that people has a lot of idea’s about this. But as
> Fokko
> > > already mentioned this is out of scope for the issue what we are trying
> > to
> > > solve. I also have some idea’s about this but I like to limit this
> PR/AIP
> > > to the webserver.
> > > >
> > > > For now my PR does solve 1 and 2 and the rest of the behaviour (like
> > > scheduling) is untouched.
> > > >
> > > > Gr,
> > > > Peter
> > > >
> > >
> > >
> >
>

Re: [DISCUSS] AIP-12 Persist DAG into DB

[Kevin Yang <yr...@gmail.com>]

Hi Julian, I'm definitely aligned with you guys on making the webserver
independent of DAG parsing, just the end goal to me would be to build a
complete story around serializing DAG--and move with the story in mind. I
feel like you guys may already have a list of dynamic features we need to
deprecate/change, if that is the case feel free to open the discussion on
what we do to them with DAG serialization.

Julian, Ash, Dan, on 2nd thought I do agree that if we can meet the
requirements Dan mentioned, it would be nice to have them stored in the DB.
Some combined solutions like having a column of serialized graph in the
serialized dag table can potentially meet all requirements. What format we
end up using to represent DAG between components is now less important
IMO--fine to refactor those endpoints only need DagModel to use only
DagModel, easy to do a batch replacement if we decide otherwise later. More
important is to define this source of truth for serialized DAG.

Ash, ty for the email list, I'll tune my filters accordingly :D I'm leaning
towards having a separate process for the parser so we got no scheduler
dependency etc for this parser but we can discuss this in another thread.

On Fri, Mar 8, 2019 at 8:57 AM Dan Davydov <dd...@twitter.com.invalid>
wrote:

> >
> > Personally I don’t understand why people are pushing for a JSON-based DAG
> > representation
>
> It sounds like you agree that DAGs should be serialized (just in the DB
> instead of JSON), so will only address why JSON is better than MySQL (AKA
> serializing at the DAG level vs the task level) as far as I can see, and
> not why we need serialization. If you zoom out and look at all the use
> cases of serialized DAGs, e.g. having the scheduler use them instead of
> parsing DAGs directly, then it becomes clear that we need all appropriate
> metadata in these DAGs, (operator params, DAG properties, etc), in which
> case it's not clear how it will fit nicely into a DB table (unless you
> wanted to do something like (parent_task_id, task_id, task_params), also
> keep in mind that we will need to store different versions of each DAG in
> the future so that we can ensure consistency in a dagrun, i.e. each task in
> a dagrun uses the same version of a DAG.
>
> I think some of our requirements should be:
> 1. The data model will lead to acceptable performance in all of its
> consumers (scheduler, webserver, workers), i.e. no n+1 access patterns (my
> biggest concern about serializing at task level as you propose vs at DAG
> level)
> 2. We can have versioning of serialized DAGs
> 3. The ability to separate DAGs into their own data store (e.g. no reliance
> on joins between the new table and the old one)
> 4. One source of truth/serialized representation for DAGs (currently we
> have SimpleDAG)
>
> If we can full-fill all of these requirements and serialize at the task
> level rather than the DAG level in the DB, then I agree that probably makes
> more sense.
>
>
> > In the proposed PR’s we (Peter, Bas and me) aim to avoid re-parsing DAG
> > files by querying all the required information from the database. In one
> or
> > two cases this may however not be possible, in which case we might either
> > have to fall back on the DAG file or add the missing information into the
> > database. We can tackle these problems as we encounter them.
>
> I think you would have the support of many of committers in removing any
> use-cases that stand in the way of full serialization, that being said if
> we need to remove features we need to do this carefully and thoughtfully,
> and ideally with proposed alternatives/work-arounds to cover the removals.
>
> The counter argument: this PR removes the need for the confusing "Refresh"
> > button from the UI, and in general you only pay the cost for the
> expensive
> > DAGs when you ask about them. (I don't know what/when we call the
> > /pickle_info endpoint of the top of my head)
>
> Probably worth splitting out into a separate thread, but I'm actually not
> sure the refresh button does anything, I think we should double check... I
> think about 2 years ago there was a commit made that made gunicorn
> webservers automatically rotate underneath flask (each one would reparse
> the DAGbag). Even if it works we should probably remove it since the
> webserver refresh interval is pretty fast, and it just causes confusion to
> users and implies that the DAGs are not refreshed automatically.
>
> Do you mean https://json5.org/ or is this a typo? That might be okay for a
> > nicer user front end, but the "canonical" version stored in the DB should
> > be something "plainer" like just JSON.
>
> I think he got this from my reply, and it was just an example, but you are
> right, I agree JSON would be better than JSON5.
>
> On Fri, Mar 8, 2019 at 8:53 AM Ash Berlin-Taylor <as...@apache.org> wrote:
>
> > Comments inline.
> >
> > > On 8 Mar 2019, at 11:28, Kevin Yang <yr...@gmail.com> wrote:
> > >
> > > Hi all,
> > > When I was preparing some work related to this AIP I found something
> > very concerning. I noticed this JIRA ticket <
> > https://issues.apache.org/jira/browse/AIRFLOW-3562> is trying to remove
> > the dependency of dagbag from webserver, which is awesome--we wanted
> badly
> > but never got to start work on. However when I looked at some subtasks of
> > it, which try to remove dagbag dependency from each endpoint, I found the
> > way we remove the dependency of dagbag is not very ideal. For example
> this
> > PR <https://github.com/apache/airflow/pull/4867/files> will require us
> to
> > parse the dag file each time we hit the endpoint.
> >
> > The counter argument: this PR removes the need for the confusing
> "Refresh"
> > button from the UI, and in general you only pay the cost for the
> expensive
> > DAGs when you ask about them. (I don't know what/when we call the
> > /pickle_info endpoint of the top of my head)
> >
> > This end point may be one to hold off on (as it can ask for multiple
> dags)
> > but there are some that def don't need a full dag bag or to even parse
> the
> > dag file, the current DAG model has enough info.
> >
> > >
> > >
> > > If we go down this path, we indeed can get rid of the dagbag dependency
> > easily, but we will have to 1. increase the DB load( not too concerning
> at
> > the moment ), 2. wait the DAG file to be parsed before getting the page
> > back, potentially multiple times. DAG file can sometimes take quite a
> while
> > to parse, e.g. we have some framework DAG files generating large number
> of
> > DAGs from some static config files or even jupyter notebooks and they can
> > take 30+ seconds to parse. Yes we don't like large DAG files but people
> do
> > see the beauty of code as config and sometimes heavily abuseleverage it.
> > Assuming all users have the same nice small python file that can be
> parsed
> > fast, I'm still a bit worried about this approach. Continuing on this
> path
> > means we've chosen DagModel to be the serialized representation of DAG
> and
> > DB columns to hold different properties--it can be one candidate but I
> > don't know if we should settle on that now. I would personally prefer a
> > more compact, e.g. JSON5, and easy to scale representation( such that
> > serializing new fields != DB upgrade).
> >
> > Do you mean https://json5.org/ or is this a typo? That might be okay for
> > a nicer user front end, but the "canonical" version stored in the DB
> should
> > be something "plainer" like just JSON.
> >
> > I'm not sure that "serializing new fields != DB upgrade" is that big of a
> > concern, as we don't add fields that often. One possible way of dealing
> > with it if we do is to have a hybrid approach - a few distinct columns,
> but
> > then a JSON blob. (and if we were only to support postgres we could just
> > use JSONb. But I think our friends at Google may object ;) )
> >
> > Adding a new column in a DB migration with a default NULL shouldn't be an
> > expensive operation, or difficult to achieve.
> >
> >
> > >
> > > In my imagination we would have to collect the list of dynamic features
> > depending on unserializable fields of a DAG and start a discussion/vote
> on
> > dropping support of them( I'm working on this but if anyone has already
> > done so please take over), decide on the serialized representation of a
> DAG
> > and then replace dagbag with it in webserver. Per previous discussion and
> > some offline discussions with Dan, one future of DAG serialization that I
> > like would look similar to this:
> > >
> >
> > > https://imgur.com/ncqqQgc
> >
> > Something I've thought about before for other things was to embed an API
> > server _into_ the scheduler - this would be useful for k8s healthchecks,
> > native Prometheus metrics without needed statsd bridge, and could have
> > endpoints to get information such as this directly.
> >
> > I was thinking it would be _in_ the scheduler process using either
> threads
> > (ick. Python's still got a GIL doesn't it?) or using async/twisted etc.
> > (not a side-car process like we have with the logs webserver for `airflow
> > worker`).
> >
> > (This is possibly an unrelated discussion, but might be worth talking
> > about?)
> >
> > > We can still discuss/vote which approach we want to take but I don't
> > want the door to above design to be shut right now or we have to spend a
> > lot effort switch path later.
> > >
> > > Bas and Peter, I'm very sorry to extend the discussion but I do think
> > this is tightly related to the AIP and PRs behind it. And my sincere
> > apology for bringing this up so late( I only pull the open PR list
> > occasionally, if there's a way to subscribe to new PR event I'd love to
> > know how).
> >
> > It's noisy, but you can subscribe to commits@airflow.apache.org (but be
> > warned, this also includes all Jira tickets, edits of every comment on
> > github etc.).
> >
> >
> > >
> > > Cheers,
> > > Kevin Y
> > >
> > > On Thu, Feb 28, 2019 at 1:36 PM Peter van t Hof <pjrvanthof@gmail.com
> > <ma...@gmail.com>> wrote:
> > > Hi all,
> > >
> > > Just some comments one the point Bolke dit give in relation of my PR.
> > >
> > > At first, the main focus is: making the webserver stateless.
> > >
> > > > 1) Make the webserver stateless: needs the graph of the *current* dag
> > >
> > > This is the main goal but for this a lot more PR’s will be coming once
> > my current is merged. For edges and graph view this is covered in my PR
> > already.
> > >
> > > > 2) Version dags: for consistency mainly and not requiring parsing of
> > the
> > > > dag on every loop
> > >
> > > In my PR the historical graphs will be stored for each DagRun. This
> > means that you can see if an older DagRun was the same graph structure,
> > even if some tasks does not exists anymore in the current graph.
> Especially
> > for dynamic DAG’s this is very useful.
> > >
> > > > 3) Make the scheduler not require DAG files. This could be done if
> the
> > > > edges contain all information when to trigger the next task. We can
> > then
> > > > have event driven dag parsing outside of the scheduler loop, ie. by
> the
> > > > cli. Storage can also be somewhere else (git, artifactory,
> filesystem,
> > > > whatever).
> > >
> > > The scheduler is almost untouched in this PR. The only thing that is
> > added is that this edges are saved to the database but the scheduling
> > itself din’t change. The scheduler depends now still on the DAG object.
> > >
> > > > 4) Fully serialise the dag so it becomes transferable to workers
> > >
> > > It nice to see that people has a lot of idea’s about this. But as Fokko
> > already mentioned this is out of scope for the issue what we are trying
> to
> > solve. I also have some idea’s about this but I like to limit this PR/AIP
> > to the webserver.
> > >
> > > For now my PR does solve 1 and 2 and the rest of the behaviour (like
> > scheduling) is untouched.
> > >
> > > Gr,
> > > Peter
> > >
> >
> >
>

Re: [DISCUSS] AIP-12 Persist DAG into DB

[Dan Davydov <dd...@twitter.com.INVALID>]

>
> Personally I don’t understand why people are pushing for a JSON-based DAG
> representation

It sounds like you agree that DAGs should be serialized (just in the DB
instead of JSON), so will only address why JSON is better than MySQL (AKA
serializing at the DAG level vs the task level) as far as I can see, and
not why we need serialization. If you zoom out and look at all the use
cases of serialized DAGs, e.g. having the scheduler use them instead of
parsing DAGs directly, then it becomes clear that we need all appropriate
metadata in these DAGs, (operator params, DAG properties, etc), in which
case it's not clear how it will fit nicely into a DB table (unless you
wanted to do something like (parent_task_id, task_id, task_params), also
keep in mind that we will need to store different versions of each DAG in
the future so that we can ensure consistency in a dagrun, i.e. each task in
a dagrun uses the same version of a DAG.

I think some of our requirements should be:
1. The data model will lead to acceptable performance in all of its
consumers (scheduler, webserver, workers), i.e. no n+1 access patterns (my
biggest concern about serializing at task level as you propose vs at DAG
level)
2. We can have versioning of serialized DAGs
3. The ability to separate DAGs into their own data store (e.g. no reliance
on joins between the new table and the old one)
4. One source of truth/serialized representation for DAGs (currently we
have SimpleDAG)

If we can full-fill all of these requirements and serialize at the task
level rather than the DAG level in the DB, then I agree that probably makes
more sense.


> In the proposed PR’s we (Peter, Bas and me) aim to avoid re-parsing DAG
> files by querying all the required information from the database. In one or
> two cases this may however not be possible, in which case we might either
> have to fall back on the DAG file or add the missing information into the
> database. We can tackle these problems as we encounter them.

I think you would have the support of many of committers in removing any
use-cases that stand in the way of full serialization, that being said if
we need to remove features we need to do this carefully and thoughtfully,
and ideally with proposed alternatives/work-arounds to cover the removals.

The counter argument: this PR removes the need for the confusing "Refresh"
> button from the UI, and in general you only pay the cost for the expensive
> DAGs when you ask about them. (I don't know what/when we call the
> /pickle_info endpoint of the top of my head)

Probably worth splitting out into a separate thread, but I'm actually not
sure the refresh button does anything, I think we should double check... I
think about 2 years ago there was a commit made that made gunicorn
webservers automatically rotate underneath flask (each one would reparse
the DAGbag). Even if it works we should probably remove it since the
webserver refresh interval is pretty fast, and it just causes confusion to
users and implies that the DAGs are not refreshed automatically.

Do you mean https://json5.org/ or is this a typo? That might be okay for a
> nicer user front end, but the "canonical" version stored in the DB should
> be something "plainer" like just JSON.

I think he got this from my reply, and it was just an example, but you are
right, I agree JSON would be better than JSON5.

On Fri, Mar 8, 2019 at 8:53 AM Ash Berlin-Taylor <as...@apache.org> wrote:

> Comments inline.
>
> > On 8 Mar 2019, at 11:28, Kevin Yang <yr...@gmail.com> wrote:
> >
> > Hi all,
> > When I was preparing some work related to this AIP I found something
> very concerning. I noticed this JIRA ticket <
> https://issues.apache.org/jira/browse/AIRFLOW-3562> is trying to remove
> the dependency of dagbag from webserver, which is awesome--we wanted badly
> but never got to start work on. However when I looked at some subtasks of
> it, which try to remove dagbag dependency from each endpoint, I found the
> way we remove the dependency of dagbag is not very ideal. For example this
> PR <https://github.com/apache/airflow/pull/4867/files> will require us to
> parse the dag file each time we hit the endpoint.
>
> The counter argument: this PR removes the need for the confusing "Refresh"
> button from the UI, and in general you only pay the cost for the expensive
> DAGs when you ask about them. (I don't know what/when we call the
> /pickle_info endpoint of the top of my head)
>
> This end point may be one to hold off on (as it can ask for multiple dags)
> but there are some that def don't need a full dag bag or to even parse the
> dag file, the current DAG model has enough info.
>
> >
> >
> > If we go down this path, we indeed can get rid of the dagbag dependency
> easily, but we will have to 1. increase the DB load( not too concerning at
> the moment ), 2. wait the DAG file to be parsed before getting the page
> back, potentially multiple times. DAG file can sometimes take quite a while
> to parse, e.g. we have some framework DAG files generating large number of
> DAGs from some static config files or even jupyter notebooks and they can
> take 30+ seconds to parse. Yes we don't like large DAG files but people do
> see the beauty of code as config and sometimes heavily abuseleverage it.
> Assuming all users have the same nice small python file that can be parsed
> fast, I'm still a bit worried about this approach. Continuing on this path
> means we've chosen DagModel to be the serialized representation of DAG and
> DB columns to hold different properties--it can be one candidate but I
> don't know if we should settle on that now. I would personally prefer a
> more compact, e.g. JSON5, and easy to scale representation( such that
> serializing new fields != DB upgrade).
>
> Do you mean https://json5.org/ or is this a typo? That might be okay for
> a nicer user front end, but the "canonical" version stored in the DB should
> be something "plainer" like just JSON.
>
> I'm not sure that "serializing new fields != DB upgrade" is that big of a
> concern, as we don't add fields that often. One possible way of dealing
> with it if we do is to have a hybrid approach - a few distinct columns, but
> then a JSON blob. (and if we were only to support postgres we could just
> use JSONb. But I think our friends at Google may object ;) )
>
> Adding a new column in a DB migration with a default NULL shouldn't be an
> expensive operation, or difficult to achieve.
>
>
> >
> > In my imagination we would have to collect the list of dynamic features
> depending on unserializable fields of a DAG and start a discussion/vote on
> dropping support of them( I'm working on this but if anyone has already
> done so please take over), decide on the serialized representation of a DAG
> and then replace dagbag with it in webserver. Per previous discussion and
> some offline discussions with Dan, one future of DAG serialization that I
> like would look similar to this:
> >
>
> > https://imgur.com/ncqqQgc
>
> Something I've thought about before for other things was to embed an API
> server _into_ the scheduler - this would be useful for k8s healthchecks,
> native Prometheus metrics without needed statsd bridge, and could have
> endpoints to get information such as this directly.
>
> I was thinking it would be _in_ the scheduler process using either threads
> (ick. Python's still got a GIL doesn't it?) or using async/twisted etc.
> (not a side-car process like we have with the logs webserver for `airflow
> worker`).
>
> (This is possibly an unrelated discussion, but might be worth talking
> about?)
>
> > We can still discuss/vote which approach we want to take but I don't
> want the door to above design to be shut right now or we have to spend a
> lot effort switch path later.
> >
> > Bas and Peter, I'm very sorry to extend the discussion but I do think
> this is tightly related to the AIP and PRs behind it. And my sincere
> apology for bringing this up so late( I only pull the open PR list
> occasionally, if there's a way to subscribe to new PR event I'd love to
> know how).
>
> It's noisy, but you can subscribe to commits@airflow.apache.org (but be
> warned, this also includes all Jira tickets, edits of every comment on
> github etc.).
>
>
> >
> > Cheers,
> > Kevin Y
> >
> > On Thu, Feb 28, 2019 at 1:36 PM Peter van t Hof <pjrvanthof@gmail.com
> <ma...@gmail.com>> wrote:
> > Hi all,
> >
> > Just some comments one the point Bolke dit give in relation of my PR.
> >
> > At first, the main focus is: making the webserver stateless.
> >
> > > 1) Make the webserver stateless: needs the graph of the *current* dag
> >
> > This is the main goal but for this a lot more PR’s will be coming once
> my current is merged. For edges and graph view this is covered in my PR
> already.
> >
> > > 2) Version dags: for consistency mainly and not requiring parsing of
> the
> > > dag on every loop
> >
> > In my PR the historical graphs will be stored for each DagRun. This
> means that you can see if an older DagRun was the same graph structure,
> even if some tasks does not exists anymore in the current graph. Especially
> for dynamic DAG’s this is very useful.
> >
> > > 3) Make the scheduler not require DAG files. This could be done if the
> > > edges contain all information when to trigger the next task. We can
> then
> > > have event driven dag parsing outside of the scheduler loop, ie. by the
> > > cli. Storage can also be somewhere else (git, artifactory, filesystem,
> > > whatever).
> >
> > The scheduler is almost untouched in this PR. The only thing that is
> added is that this edges are saved to the database but the scheduling
> itself din’t change. The scheduler depends now still on the DAG object.
> >
> > > 4) Fully serialise the dag so it becomes transferable to workers
> >
> > It nice to see that people has a lot of idea’s about this. But as Fokko
> already mentioned this is out of scope for the issue what we are trying to
> solve. I also have some idea’s about this but I like to limit this PR/AIP
> to the webserver.
> >
> > For now my PR does solve 1 and 2 and the rest of the behaviour (like
> scheduling) is untouched.
> >
> > Gr,
> > Peter
> >
>
>

Re: [DISCUSS] AIP-12 Persist DAG into DB

[Ash Berlin-Taylor <as...@apache.org>]

Comments inline.

> On 8 Mar 2019, at 11:28, Kevin Yang <yr...@gmail.com> wrote:
> 
> Hi all,
> When I was preparing some work related to this AIP I found something very concerning. I noticed this JIRA ticket <https://issues.apache.org/jira/browse/AIRFLOW-3562> is trying to remove the dependency of dagbag from webserver, which is awesome--we wanted badly but never got to start work on. However when I looked at some subtasks of it, which try to remove dagbag dependency from each endpoint, I found the way we remove the dependency of dagbag is not very ideal. For example this PR <https://github.com/apache/airflow/pull/4867/files> will require us to parse the dag file each time we hit the endpoint.

The counter argument: this PR removes the need for the confusing "Refresh" button from the UI, and in general you only pay the cost for the expensive DAGs when you ask about them. (I don't know what/when we call the /pickle_info endpoint of the top of my head)

This end point may be one to hold off on (as it can ask for multiple dags) but there are some that def don't need a full dag bag or to even parse the dag file, the current DAG model has enough info.

>  
> 
> If we go down this path, we indeed can get rid of the dagbag dependency easily, but we will have to 1. increase the DB load( not too concerning at the moment ), 2. wait the DAG file to be parsed before getting the page back, potentially multiple times. DAG file can sometimes take quite a while to parse, e.g. we have some framework DAG files generating large number of DAGs from some static config files or even jupyter notebooks and they can take 30+ seconds to parse. Yes we don't like large DAG files but people do see the beauty of code as config and sometimes heavily abuseleverage it. Assuming all users have the same nice small python file that can be parsed fast, I'm still a bit worried about this approach. Continuing on this path means we've chosen DagModel to be the serialized representation of DAG and DB columns to hold different properties--it can be one candidate but I don't know if we should settle on that now. I would personally prefer a more compact, e.g. JSON5, and easy to scale representation( such that serializing new fields != DB upgrade). 

Do you mean https://json5.org/ or is this a typo? That might be okay for a nicer user front end, but the "canonical" version stored in the DB should be something "plainer" like just JSON.

I'm not sure that "serializing new fields != DB upgrade" is that big of a concern, as we don't add fields that often. One possible way of dealing with it if we do is to have a hybrid approach - a few distinct columns, but then a JSON blob. (and if we were only to support postgres we could just use JSONb. But I think our friends at Google may object ;) )

Adding a new column in a DB migration with a default NULL shouldn't be an expensive operation, or difficult to achieve.


> 
> In my imagination we would have to collect the list of dynamic features depending on unserializable fields of a DAG and start a discussion/vote on dropping support of them( I'm working on this but if anyone has already done so please take over), decide on the serialized representation of a DAG and then replace dagbag with it in webserver. Per previous discussion and some offline discussions with Dan, one future of DAG serialization that I like would look similar to this:
> 

> https://imgur.com/ncqqQgc

Something I've thought about before for other things was to embed an API server _into_ the scheduler - this would be useful for k8s healthchecks, native Prometheus metrics without needed statsd bridge, and could have endpoints to get information such as this directly. 

I was thinking it would be _in_ the scheduler process using either threads (ick. Python's still got a GIL doesn't it?) or using async/twisted etc. (not a side-car process like we have with the logs webserver for `airflow worker`).

(This is possibly an unrelated discussion, but might be worth talking about?)

> We can still discuss/vote which approach we want to take but I don't want the door to above design to be shut right now or we have to spend a lot effort switch path later.
> 
> Bas and Peter, I'm very sorry to extend the discussion but I do think this is tightly related to the AIP and PRs behind it. And my sincere apology for bringing this up so late( I only pull the open PR list occasionally, if there's a way to subscribe to new PR event I'd love to know how).

It's noisy, but you can subscribe to commits@airflow.apache.org (but be warned, this also includes all Jira tickets, edits of every comment on github etc.).


> 
> Cheers,
> Kevin Y
> 
> On Thu, Feb 28, 2019 at 1:36 PM Peter van t Hof <pjrvanthof@gmail.com <ma...@gmail.com>> wrote:
> Hi all,
> 
> Just some comments one the point Bolke dit give in relation of my PR.
> 
> At first, the main focus is: making the webserver stateless. 
> 
> > 1) Make the webserver stateless: needs the graph of the *current* dag
> 
> This is the main goal but for this a lot more PR’s will be coming once my current is merged. For edges and graph view this is covered in my PR already.
> 
> > 2) Version dags: for consistency mainly and not requiring parsing of the
> > dag on every loop
> 
> In my PR the historical graphs will be stored for each DagRun. This means that you can see if an older DagRun was the same graph structure, even if some tasks does not exists anymore in the current graph. Especially for dynamic DAG’s this is very useful.
> 
> > 3) Make the scheduler not require DAG files. This could be done if the
> > edges contain all information when to trigger the next task. We can then
> > have event driven dag parsing outside of the scheduler loop, ie. by the
> > cli. Storage can also be somewhere else (git, artifactory, filesystem,
> > whatever).
> 
> The scheduler is almost untouched in this PR. The only thing that is added is that this edges are saved to the database but the scheduling itself din’t change. The scheduler depends now still on the DAG object.
> 
> > 4) Fully serialise the dag so it becomes transferable to workers
> 
> It nice to see that people has a lot of idea’s about this. But as Fokko already mentioned this is out of scope for the issue what we are trying to solve. I also have some idea’s about this but I like to limit this PR/AIP to the webserver.
> 
> For now my PR does solve 1 and 2 and the rest of the behaviour (like scheduling) is untouched.
> 
> Gr,
> Peter
> 

Re: [DISCUSS] AIP-12 Persist DAG into DB

[Kevin Yang <yr...@gmail.com>]

Ty Xiangdong, my bad there. Attached the file to this email and also
uploaded it here <https://photos.app.goo.gl/Rr5BsHvxXEXnbY5K7> and here
<https://imgur.com/ncqqQgc>.

Cheers,
Kevin Y

On Fri, Mar 8, 2019 at 3:42 AM Deng Xiaodong <xd...@gmail.com> wrote:

> Hi Kevin,
>
> The image you attached is not displayed properly. May you consider
> uploading it somewhere then provide a link instead?
>
> Thanks!
>
> XD
>
> On Fri, Mar 8, 2019 at 19:38 Kevin Yang <yr...@gmail.com> wrote:
>
> > Hi all,
> > When I was preparing some work related to this AIP I found something very
> > concerning. I noticed this JIRA ticket
> > <https://issues.apache.org/jira/browse/AIRFLOW-3562> is trying to remove
> > the dependency of dagbag from webserver, which is awesome--we wanted
> badly
> > but never got to start work on. However when I looked at some subtasks of
> > it, which try to remove dagbag dependency from each endpoint, I found the
> > way we remove the dependency of dagbag is not very ideal. For example
> this
> > PR <https://github.com/apache/airflow/pull/4867/files> will require us
> to
> > parse the dag file each time we hit the endpoint.
> >
> > If we go down this path, we indeed can get rid of the dagbag dependency
> > easily, but we will have to 1. increase the DB load( not too concerning
> at
> > the moment ), 2. wait the DAG file to be parsed before getting the page
> > back, potentially multiple times. DAG file can sometimes take quite a
> while
> > to parse, e.g. we have some framework DAG files generating large number
> of
> > DAGs from some static config files or even jupyter notebooks and they can
> > take 30+ seconds to parse. Yes we don't like large DAG files but people
> do
> > see the beauty of code as config and sometimes heavily abuseleverage it.
> > Assuming all users have the same nice small python file that can be
> parsed
> > fast, I'm still a bit worried about this approach. Continuing on this
> path
> > means we've chosen DagModel to be the serialized representation of DAG
> and
> > DB columns to hold different properties--it can be one candidate but I
> > don't know if we should settle on that now. I would personally prefer a
> > more compact, e.g. JSON5, and easy to scale representation( such that
> > serializing new fields != DB upgrade).
> >
> > In my imagination we would have to collect the list of dynamic features
> > depending on unserializable fields of a DAG and start a discussion/vote
> on
> > dropping support of them( I'm working on this but if anyone has already
> > done so please take over), decide on the serialized representation of a
> DAG
> > and then replace dagbag with it in webserver. Per previous discussion and
> > some offline discussions with Dan, one future of DAG serialization that I
> > like would look similar to this:
> > [image: airflow_new_arch.jpg]
> > We can still discuss/vote which approach we want to take but I don't want
> > the door to above design to be shut right now or we have to spend a lot
> > effort switch path later.
> >
> > Bas and Peter, I'm very sorry to extend the discussion but I do think
> this
> > is tightly related to the AIP and PRs behind it. And my sincere apology
> for
> > bringing this up so late( I only pull the open PR list occasionally, if
> > there's a way to subscribe to new PR event I'd love to know how).
> >
> > Cheers,
> > Kevin Y
> >
> > On Thu, Feb 28, 2019 at 1:36 PM Peter van t Hof <pj...@gmail.com>
> > wrote:
> >
> >> Hi all,
> >>
> >> Just some comments one the point Bolke dit give in relation of my PR.
> >>
> >> At first, the main focus is: making the webserver stateless.
> >>
> >> > 1) Make the webserver stateless: needs the graph of the *current* dag
> >>
> >> This is the main goal but for this a lot more PR’s will be coming once
> my
> >> current is merged. For edges and graph view this is covered in my PR
> >> already.
> >>
> >> > 2) Version dags: for consistency mainly and not requiring parsing of
> the
> >> > dag on every loop
> >>
> >> In my PR the historical graphs will be stored for each DagRun. This
> means
> >> that you can see if an older DagRun was the same graph structure, even
> if
> >> some tasks does not exists anymore in the current graph. Especially for
> >> dynamic DAG’s this is very useful.
> >>
> >> > 3) Make the scheduler not require DAG files. This could be done if the
> >> > edges contain all information when to trigger the next task. We can
> then
> >> > have event driven dag parsing outside of the scheduler loop, ie. by
> the
> >> > cli. Storage can also be somewhere else (git, artifactory, filesystem,
> >> > whatever).
> >>
> >> The scheduler is almost untouched in this PR. The only thing that is
> >> added is that this edges are saved to the database but the scheduling
> >> itself din’t change. The scheduler depends now still on the DAG object.
> >>
> >> > 4) Fully serialise the dag so it becomes transferable to workers
> >>
> >> It nice to see that people has a lot of idea’s about this. But as Fokko
> >> already mentioned this is out of scope for the issue what we are trying
> to
> >> solve. I also have some idea’s about this but I like to limit this
> PR/AIP
> >> to the webserver.
> >>
> >> For now my PR does solve 1 and 2 and the rest of the behaviour (like
> >> scheduling) is untouched.
> >>
> >> Gr,
> >> Peter
> >>
> >>
>

Re: [DISCUSS] AIP-12 Persist DAG into DB

[Deng Xiaodong <xd...@gmail.com>]

Hi Kevin,

The image you attached is not displayed properly. May you consider
uploading it somewhere then provide a link instead?

Thanks!

XD

On Fri, Mar 8, 2019 at 19:38 Kevin Yang <yr...@gmail.com> wrote:

> Hi all,
> When I was preparing some work related to this AIP I found something very
> concerning. I noticed this JIRA ticket
> <https://issues.apache.org/jira/browse/AIRFLOW-3562> is trying to remove
> the dependency of dagbag from webserver, which is awesome--we wanted badly
> but never got to start work on. However when I looked at some subtasks of
> it, which try to remove dagbag dependency from each endpoint, I found the
> way we remove the dependency of dagbag is not very ideal. For example this
> PR <https://github.com/apache/airflow/pull/4867/files> will require us to
> parse the dag file each time we hit the endpoint.
>
> If we go down this path, we indeed can get rid of the dagbag dependency
> easily, but we will have to 1. increase the DB load( not too concerning at
> the moment ), 2. wait the DAG file to be parsed before getting the page
> back, potentially multiple times. DAG file can sometimes take quite a while
> to parse, e.g. we have some framework DAG files generating large number of
> DAGs from some static config files or even jupyter notebooks and they can
> take 30+ seconds to parse. Yes we don't like large DAG files but people do
> see the beauty of code as config and sometimes heavily abuseleverage it.
> Assuming all users have the same nice small python file that can be parsed
> fast, I'm still a bit worried about this approach. Continuing on this path
> means we've chosen DagModel to be the serialized representation of DAG and
> DB columns to hold different properties--it can be one candidate but I
> don't know if we should settle on that now. I would personally prefer a
> more compact, e.g. JSON5, and easy to scale representation( such that
> serializing new fields != DB upgrade).
>
> In my imagination we would have to collect the list of dynamic features
> depending on unserializable fields of a DAG and start a discussion/vote on
> dropping support of them( I'm working on this but if anyone has already
> done so please take over), decide on the serialized representation of a DAG
> and then replace dagbag with it in webserver. Per previous discussion and
> some offline discussions with Dan, one future of DAG serialization that I
> like would look similar to this:
> [image: airflow_new_arch.jpg]
> We can still discuss/vote which approach we want to take but I don't want
> the door to above design to be shut right now or we have to spend a lot
> effort switch path later.
>
> Bas and Peter, I'm very sorry to extend the discussion but I do think this
> is tightly related to the AIP and PRs behind it. And my sincere apology for
> bringing this up so late( I only pull the open PR list occasionally, if
> there's a way to subscribe to new PR event I'd love to know how).
>
> Cheers,
> Kevin Y
>
> On Thu, Feb 28, 2019 at 1:36 PM Peter van t Hof <pj...@gmail.com>
> wrote:
>
>> Hi all,
>>
>> Just some comments one the point Bolke dit give in relation of my PR.
>>
>> At first, the main focus is: making the webserver stateless.
>>
>> > 1) Make the webserver stateless: needs the graph of the *current* dag
>>
>> This is the main goal but for this a lot more PR’s will be coming once my
>> current is merged. For edges and graph view this is covered in my PR
>> already.
>>
>> > 2) Version dags: for consistency mainly and not requiring parsing of the
>> > dag on every loop
>>
>> In my PR the historical graphs will be stored for each DagRun. This means
>> that you can see if an older DagRun was the same graph structure, even if
>> some tasks does not exists anymore in the current graph. Especially for
>> dynamic DAG’s this is very useful.
>>
>> > 3) Make the scheduler not require DAG files. This could be done if the
>> > edges contain all information when to trigger the next task. We can then
>> > have event driven dag parsing outside of the scheduler loop, ie. by the
>> > cli. Storage can also be somewhere else (git, artifactory, filesystem,
>> > whatever).
>>
>> The scheduler is almost untouched in this PR. The only thing that is
>> added is that this edges are saved to the database but the scheduling
>> itself din’t change. The scheduler depends now still on the DAG object.
>>
>> > 4) Fully serialise the dag so it becomes transferable to workers
>>
>> It nice to see that people has a lot of idea’s about this. But as Fokko
>> already mentioned this is out of scope for the issue what we are trying to
>> solve. I also have some idea’s about this but I like to limit this PR/AIP
>> to the webserver.
>>
>> For now my PR does solve 1 and 2 and the rest of the behaviour (like
>> scheduling) is untouched.
>>
>> Gr,
>> Peter
>>
>>

Re: [DISCUSS] AIP-12 Persist DAG into DB

[Julian De Ruiter <ju...@godatadriven.com>]

Hi all,

Personally I don’t understand why people are pushing for a JSON-based DAG representation. For serializing DAGs between processes it may make some sense, if that allows us to decouple the different components from the main database. However, that still requires some (large!) architectural changes to the status quo. However, w.r.t. to the database representation of DAGs, I would much rather have a table structure with well-defined fields in the database than some serialized JSON field with a potential minefield of different fields. Switching to JSON is simply moving the problem elsewhere.

In the proposed PR’s we (Peter, Bas and me) aim to avoid re-parsing DAG files by querying all the required information from the database. In one or two cases this may however not be possible, in which case we might either have to fall back on the DAG file or add the missing information into the database. We can tackle these problems as we encounter them.

The end goal is to make the webserver independent of DAG parsing.

Best regards / met vriendelijke groet,

Julian de Ruiter

From: Kevin Yang <yr...@gmail.com>
Reply-To: "dev@airflow.apache.org" <de...@airflow.apache.org>
Date: Friday, 8 March 2019 at 12:38
To: "dev@airflow.apache.org" <de...@airflow.apache.org>
Subject: Re: [DISCUSS] AIP-12 Persist DAG into DB

Hi all,
When I was preparing some work related to this AIP I found something very concerning. I noticed this JIRA ticket<https://issues.apache.org/jira/browse/AIRFLOW-3562> is trying to remove the dependency of dagbag from webserver, which is awesome--we wanted badly but never got to start work on. However when I looked at some subtasks of it, which try to remove dagbag dependency from each endpoint, I found the way we remove the dependency of dagbag is not very ideal. For example this PR<https://github.com/apache/airflow/pull/4867/files> will require us to parse the dag file each time we hit the endpoint.

If we go down this path, we indeed can get rid of the dagbag dependency easily, but we will have to 1. increase the DB load( not too concerning at the moment ), 2. wait the DAG file to be parsed before getting the page back, potentially multiple times. DAG file can sometimes take quite a while to parse, e.g. we have some framework DAG files generating large number of DAGs from some static config files or even jupyter notebooks and they can take 30+ seconds to parse. Yes we don't like large DAG files but people do see the beauty of code as config and sometimes heavily abuseleverage it. Assuming all users have the same nice small python file that can be parsed fast, I'm still a bit worried about this approach. Continuing on this path means we've chosen DagModel to be the serialized representation of DAG and DB columns to hold different properties--it can be one candidate but I don't know if we should settle on that now. I would personally prefer a more compact, e.g. JSON5, and easy to scale representation( such that serializing new fields != DB upgrade).

In my imagination we would have to collect the list of dynamic features depending on unserializable fields of a DAG and start a discussion/vote on dropping support of them( I'm working on this but if anyone has already done so please take over), decide on the serialized representation of a DAG and then replace dagbag with it in webserver. Per previous discussion and some offline discussions with Dan, one future of DAG serialization that I like would look similar to this:
[airflow_new_arch.jpg]
We can still discuss/vote which approach we want to take but I don't want the door to above design to be shut right now or we have to spend a lot effort switch path later.

Bas and Peter, I'm very sorry to extend the discussion but I do think this is tightly related to the AIP and PRs behind it. And my sincere apology for bringing this up so late( I only pull the open PR list occasionally, if there's a way to subscribe to new PR event I'd love to know how).

Cheers,
Kevin Y

On Thu, Feb 28, 2019 at 1:36 PM Peter van t Hof <pj...@gmail.com>> wrote:
Hi all,

Just some comments one the point Bolke dit give in relation of my PR.

At first, the main focus is: making the webserver stateless.

> 1) Make the webserver stateless: needs the graph of the *current* dag

This is the main goal but for this a lot more PR’s will be coming once my current is merged. For edges and graph view this is covered in my PR already.

> 2) Version dags: for consistency mainly and not requiring parsing of the
> dag on every loop

In my PR the historical graphs will be stored for each DagRun. This means that you can see if an older DagRun was the same graph structure, even if some tasks does not exists anymore in the current graph. Especially for dynamic DAG’s this is very useful.

> 3) Make the scheduler not require DAG files. This could be done if the
> edges contain all information when to trigger the next task. We can then
> have event driven dag parsing outside of the scheduler loop, ie. by the
> cli. Storage can also be somewhere else (git, artifactory, filesystem,
> whatever).

The scheduler is almost untouched in this PR. The only thing that is added is that this edges are saved to the database but the scheduling itself din’t change. The scheduler depends now still on the DAG object.

> 4) Fully serialise the dag so it becomes transferable to workers

It nice to see that people has a lot of idea’s about this. But as Fokko already mentioned this is out of scope for the issue what we are trying to solve. I also have some idea’s about this but I like to limit this PR/AIP to the webserver.

For now my PR does solve 1 and 2 and the rest of the behaviour (like scheduling) is untouched.

Gr,
Peter