Re: How to add hooks for strong deployment consistency?
I'm currently looking into building a dynamic DAG that will load user-provided data daily, into user-generated DB tables, using provided schema definitions. There will be some ordering/dependencies such that certain datasets depend on others. Since users can add new datasets at any time, both the number of nodes and their growth from day to day is unbounded. The way I am thinking of approaching this is to build an "execution plan" in advance and distribute that as an artifact, instead of having each Airflow worker re-evaluate the DAG and re-query the database of user datasets. That's doable right now by shipping a static artifact, but there are many warts. I could definitely see this as something that fits well with the described Airflow "DAGFetcher" abstraction. My wish list for an overall system based on that concept would look like this: - The DAGFetcher API is aware of execution dates, so it can query the DAG generator service with reference to a time period (not just "right now"). - The DAGFetcher API is aware of previous outputs, whether or not the DAG generator service is. This isn't a cache (performance optimization, records are ephemeral), but a ledger (immutable point in time record). For some services, like git, the service is also the ledger. In my case, the DAGRun is the product of the current git commit (versioned) and the point in time view of a database (unversioned). - I can view a "pending" DAGRun in the UI, representing a DAGRun that is expected but not yet evaluated by the DAGFetcher. - I can view a "exception" DAGRun in the UI, representing a DAGRun where the DAGFetcher raised an exception, and retry fetching through the UI. - I can alert if a DAGRun is in "pending" state for too long, or enters "exception". - Multiple DAGRuns can reference the same DAGFetcher result, if there's been no changes from day to day. - The UI can represent multiple DAGFetcher results for the same DAG, such as showing a blank entry for execution dates where a task didn't exist (it was there historically but was removed, or it's new) On Mon, Mar 5, 2018 at 4:54 PM, Maxime Beauchemin < maximebeauche...@gmail.com> wrote: > A few notes about the pickling topic to answer William: > * first reason why we cannot use pickles: jinja template objects are not > picklable, there's a hack to pickle the content of the template instead of > the object, but that breaks jinja inheritance and imports > * pickles are messy, and Airflow allows users to attach objects to DAG > objects (on_error_callable, on_retry_callable, params, ...) and pickles > will go down the recursive rabbit hole and import everything large chunks > of what's in `sys.modules` sometimes. (probably could be mitigated) > * pickles are a messy serialization format, lots of drawbacks, security > issues, incompatibility between py2 and py3, ... > * tpickles have a bad reputation, many people advised avoiding it like > plague since the feature was first built > * original approach of pickling to the db is kind of a hack > > I also agree that caching is probably required especially around large DAGs > and for "semi-stateless" web servers to operate properly. > > Max > > > On Thu, Mar 1, 2018 at 1:15 PM, David Capwell wrote: > > > We need two versions but most likely would not use either... That being > > artifactory and git (would really love for this to be pluggable!) > > > > We have our own dag fetch logic which right now pulls from git, caches, > > then redirect airflow to that directory. For us we have airflow > automated > > so you push a button to get a cluster, for this reason there are enough > > instances that we have DDOS attacked git (opps). > > > > We are planning to change this to fetch from artifactory, and have a > > stateful proxy for each cluster so we stop DDOS attacking core > > infrastructure. > > > > On Mar 1, 2018 11:45 AM, "William Wong" wrote: > > > > Also relatively new to Airflow here. Same as David above, Option 1 is not > > an option for us either for the same reasons. > > > > What I would like to see is that it can be user selectable / modifiable. > > > > Use Case: > > We have a DAG with thousands of task dependencies/tasks. After 24hrs of > > progressing, we need to take a subset of those tasks and rerun them with > a > > different configuration (reasons range from incorrect parameters to > > infrastructure issues, doesn't really matter here). > > > > What I hope can happen: > > 1. Pause DAG > > 2. Upload and tag newest dag version > > 3. Set dag_run to use latest tag, > > 4. Resolve DAG sync using clearly > > defined/documented> > > 5. Unpause DAG > > > > I do like the DagFetcher idea. This logic should shim in nicely in the > > DagBag code. Maxime, I also vote for the GitDagFetcher. Two thoughts > about > > the GitDagFetcher: > > - I probably won't use fuse across 100's of nodes in my k8s/swarm. Not > sure > > how this would work without too much trouble. > > - It might be confusing if some git sha's have no changes to
Re: How to add hooks for strong deployment consistency?
A few notes about the pickling topic to answer William: * first reason why we cannot use pickles: jinja template objects are not picklable, there's a hack to pickle the content of the template instead of the object, but that breaks jinja inheritance and imports * pickles are messy, and Airflow allows users to attach objects to DAG objects (on_error_callable, on_retry_callable, params, ...) and pickles will go down the recursive rabbit hole and import everything large chunks of what's in `sys.modules` sometimes. (probably could be mitigated) * pickles are a messy serialization format, lots of drawbacks, security issues, incompatibility between py2 and py3, ... * tpickles have a bad reputation, many people advised avoiding it like plague since the feature was first built * original approach of pickling to the db is kind of a hack I also agree that caching is probably required especially around large DAGs and for "semi-stateless" web servers to operate properly. Max On Thu, Mar 1, 2018 at 1:15 PM, David Capwell wrote: > We need two versions but most likely would not use either... That being > artifactory and git (would really love for this to be pluggable!) > > We have our own dag fetch logic which right now pulls from git, caches, > then redirect airflow to that directory. For us we have airflow automated > so you push a button to get a cluster, for this reason there are enough > instances that we have DDOS attacked git (opps). > > We are planning to change this to fetch from artifactory, and have a > stateful proxy for each cluster so we stop DDOS attacking core > infrastructure. > > On Mar 1, 2018 11:45 AM, "William Wong" wrote: > > Also relatively new to Airflow here. Same as David above, Option 1 is not > an option for us either for the same reasons. > > What I would like to see is that it can be user selectable / modifiable. > > Use Case: > We have a DAG with thousands of task dependencies/tasks. After 24hrs of > progressing, we need to take a subset of those tasks and rerun them with a > different configuration (reasons range from incorrect parameters to > infrastructure issues, doesn't really matter here). > > What I hope can happen: > 1. Pause DAG > 2. Upload and tag newest dag version > 3. Set dag_run to use latest tag, > 4. Resolve DAG sync using defined/documented> > 5. Unpause DAG > > I do like the DagFetcher idea. This logic should shim in nicely in the > DagBag code. Maxime, I also vote for the GitDagFetcher. Two thoughts about > the GitDagFetcher: > - I probably won't use fuse across 100's of nodes in my k8s/swarm. Not sure > how this would work without too much trouble. > - It might be confusing if some git sha's have no changes to a Dag. all > existing runs will be marked as outdated? probably better than nothing > anyway. > > I also vote to have some form of sort of caching behavior. I prefer not to > read in DAGs all the time. i.e. from the webserver, scheduler, *and* all > workers before starting any task over and over again. This is because, > unfortunately, the assumption that a DAG only takes seconds to load does > not hold true for large dags. With only 10k tasks within a DAG it's already > on the order of minutes. This would be untenable as we scale up to even > larger tags. (Though, I'm testing a fix for this so maybe this might not > actually be an issue anymore) > > FWIW, it seems to me that the DagPickle feature (which, for the love me I > can't seem to get it to work, no wonder it's being deprecated) would have > solved a lot of these issues fairly easily. Something along the lines of > adding pickle_id to dag_run should at help the scheduler identify the DAG > version to load and queue. But I'm not sure if it can delete out of sync > task instances. > > Lastly, sorry for the brain dump and derailing the topic, for the workers, > it seems that importing/loading in the DAG just to execute a single task is > a bit overkill isn't it? If we kept a caching feature (i.e. pickling), > perhaps we can simply cache the task and not worry about the rest of the > DAG tasks? > > Will > > On Thu, Mar 1, 2018 at 11:30 AM, Maxime Beauchemin < > > maximebeauche...@gmail.com> wrote: > > > I'm curious to hear which DagFetcher abstraction people would build or > want > > to use. > > > > So far it sounded like the most popular and flexible approach would be a > > `GitDagFetcher` where all SHAs and refs become a possibility, as opposed > to > > say a TarballOnS3DagFetcher which would require more manual artifact > > management and versioning, which represent additional [human] workflow on > > top of the already existing git-based workflow. > > > > One way I've seen this done before is by using this Git fuse (file system > > in user space) hack that creates a virtual filesystem where all SHAs and > > refs in the Git repo are exposed as a subfolder, and under each ref > > subfolder the whole repo sits as of that ref. Of course all the files are > > virtual and fetched at access time by th
Re: How to add hooks for strong deployment consistency?
We need two versions but most likely would not use either... That being artifactory and git (would really love for this to be pluggable!) We have our own dag fetch logic which right now pulls from git, caches, then redirect airflow to that directory. For us we have airflow automated so you push a button to get a cluster, for this reason there are enough instances that we have DDOS attacked git (opps). We are planning to change this to fetch from artifactory, and have a stateful proxy for each cluster so we stop DDOS attacking core infrastructure. On Mar 1, 2018 11:45 AM, "William Wong" wrote: Also relatively new to Airflow here. Same as David above, Option 1 is not an option for us either for the same reasons. What I would like to see is that it can be user selectable / modifiable. Use Case: We have a DAG with thousands of task dependencies/tasks. After 24hrs of progressing, we need to take a subset of those tasks and rerun them with a different configuration (reasons range from incorrect parameters to infrastructure issues, doesn't really matter here). What I hope can happen: 1. Pause DAG 2. Upload and tag newest dag version 3. Set dag_run to use latest tag, 4. Resolve DAG sync using 5. Unpause DAG I do like the DagFetcher idea. This logic should shim in nicely in the DagBag code. Maxime, I also vote for the GitDagFetcher. Two thoughts about the GitDagFetcher: - I probably won't use fuse across 100's of nodes in my k8s/swarm. Not sure how this would work without too much trouble. - It might be confusing if some git sha's have no changes to a Dag. all existing runs will be marked as outdated? probably better than nothing anyway. I also vote to have some form of sort of caching behavior. I prefer not to read in DAGs all the time. i.e. from the webserver, scheduler, *and* all workers before starting any task over and over again. This is because, unfortunately, the assumption that a DAG only takes seconds to load does not hold true for large dags. With only 10k tasks within a DAG it's already on the order of minutes. This would be untenable as we scale up to even larger tags. (Though, I'm testing a fix for this so maybe this might not actually be an issue anymore) FWIW, it seems to me that the DagPickle feature (which, for the love me I can't seem to get it to work, no wonder it's being deprecated) would have solved a lot of these issues fairly easily. Something along the lines of adding pickle_id to dag_run should at help the scheduler identify the DAG version to load and queue. But I'm not sure if it can delete out of sync task instances. Lastly, sorry for the brain dump and derailing the topic, for the workers, it seems that importing/loading in the DAG just to execute a single task is a bit overkill isn't it? If we kept a caching feature (i.e. pickling), perhaps we can simply cache the task and not worry about the rest of the DAG tasks? Will On Thu, Mar 1, 2018 at 11:30 AM, Maxime Beauchemin < maximebeauche...@gmail.com> wrote: > I'm curious to hear which DagFetcher abstraction people would build or want > to use. > > So far it sounded like the most popular and flexible approach would be a > `GitDagFetcher` where all SHAs and refs become a possibility, as opposed to > say a TarballOnS3DagFetcher which would require more manual artifact > management and versioning, which represent additional [human] workflow on > top of the already existing git-based workflow. > > One way I've seen this done before is by using this Git fuse (file system > in user space) hack that creates a virtual filesystem where all SHAs and > refs in the Git repo are exposed as a subfolder, and under each ref > subfolder the whole repo sits as of that ref. Of course all the files are > virtual and fetched at access time by the virtual filesystem using the git > api. So if you simply point the DagBag loader to the right [virtual] > directory, it will import the right version of the DAG. In the git world, > the alternative to that is managing temp folders and doing shallow clones > which seems like much more of a headache. Note that one tradeoff is that if > git and whatever it depends has then a need to be highly available. > > Max > > On Wed, Feb 28, 2018 at 6:55 PM, David Capwell wrote: > > > Thanks for all the details! With a pluggable fetcher we would be able to > > add our own logic for how to fetch so sounds like a good place to start > for > > something like this! > > > > On Wed, Feb 28, 2018, 4:39 PM Joy Gao wrote: > > > > > +1 on DagFetcher abstraction, very airflow-esque :) > > > > > > On Wed, Feb 28, 2018 at 11:25 AM, Maxime Beauchemin > > > wrote: > > > > Addressing a few of your questions / concerns: > > > > > > > > * The scheduler uses a multiprocess queue to queue up tasks, each > > > > subprocess is in charge of a single DAG "scheduler cycle" which > > triggers > > > > what it can for active DagRuns. Currently it fills the DagBag from > the > > > > local file system, looking for a specific module
Re: How to add hooks for strong deployment consistency?
Also relatively new to Airflow here. Same as David above, Option 1 is not an option for us either for the same reasons. What I would like to see is that it can be user selectable / modifiable. Use Case: We have a DAG with thousands of task dependencies/tasks. After 24hrs of progressing, we need to take a subset of those tasks and rerun them with a different configuration (reasons range from incorrect parameters to infrastructure issues, doesn't really matter here). What I hope can happen: 1. Pause DAG 2. Upload and tag newest dag version 3. Set dag_run to use latest tag, 4. Resolve DAG sync using 5. Unpause DAG I do like the DagFetcher idea. This logic should shim in nicely in the DagBag code. Maxime, I also vote for the GitDagFetcher. Two thoughts about the GitDagFetcher: - I probably won't use fuse across 100's of nodes in my k8s/swarm. Not sure how this would work without too much trouble. - It might be confusing if some git sha's have no changes to a Dag. all existing runs will be marked as outdated? probably better than nothing anyway. I also vote to have some form of sort of caching behavior. I prefer not to read in DAGs all the time. i.e. from the webserver, scheduler, *and* all workers before starting any task over and over again. This is because, unfortunately, the assumption that a DAG only takes seconds to load does not hold true for large dags. With only 10k tasks within a DAG it's already on the order of minutes. This would be untenable as we scale up to even larger tags. (Though, I'm testing a fix for this so maybe this might not actually be an issue anymore) FWIW, it seems to me that the DagPickle feature (which, for the love me I can't seem to get it to work, no wonder it's being deprecated) would have solved a lot of these issues fairly easily. Something along the lines of adding pickle_id to dag_run should at help the scheduler identify the DAG version to load and queue. But I'm not sure if it can delete out of sync task instances. Lastly, sorry for the brain dump and derailing the topic, for the workers, it seems that importing/loading in the DAG just to execute a single task is a bit overkill isn't it? If we kept a caching feature (i.e. pickling), perhaps we can simply cache the task and not worry about the rest of the DAG tasks? Will On Thu, Mar 1, 2018 at 11:30 AM, Maxime Beauchemin < maximebeauche...@gmail.com> wrote: > I'm curious to hear which DagFetcher abstraction people would build or want > to use. > > So far it sounded like the most popular and flexible approach would be a > `GitDagFetcher` where all SHAs and refs become a possibility, as opposed to > say a TarballOnS3DagFetcher which would require more manual artifact > management and versioning, which represent additional [human] workflow on > top of the already existing git-based workflow. > > One way I've seen this done before is by using this Git fuse (file system > in user space) hack that creates a virtual filesystem where all SHAs and > refs in the Git repo are exposed as a subfolder, and under each ref > subfolder the whole repo sits as of that ref. Of course all the files are > virtual and fetched at access time by the virtual filesystem using the git > api. So if you simply point the DagBag loader to the right [virtual] > directory, it will import the right version of the DAG. In the git world, > the alternative to that is managing temp folders and doing shallow clones > which seems like much more of a headache. Note that one tradeoff is that if > git and whatever it depends has then a need to be highly available. > > Max > > On Wed, Feb 28, 2018 at 6:55 PM, David Capwell wrote: > > > Thanks for all the details! With a pluggable fetcher we would be able to > > add our own logic for how to fetch so sounds like a good place to start > for > > something like this! > > > > On Wed, Feb 28, 2018, 4:39 PM Joy Gao wrote: > > > > > +1 on DagFetcher abstraction, very airflow-esque :) > > > > > > On Wed, Feb 28, 2018 at 11:25 AM, Maxime Beauchemin > > > wrote: > > > > Addressing a few of your questions / concerns: > > > > > > > > * The scheduler uses a multiprocess queue to queue up tasks, each > > > > subprocess is in charge of a single DAG "scheduler cycle" which > > triggers > > > > what it can for active DagRuns. Currently it fills the DagBag from > the > > > > local file system, looking for a specific module where the master > > process > > > > last saw that DAG. Fetching the DAG is a metadata operation, DAG > > > artifacts > > > > shouldn't be too large, we can assume that it takes seconds at most > to > > > > fetch a DAG, which is ok. We generally assume that the scheduler > should > > > > fully cycle every minute or so. Version-aware DagFetcher could also > > > > implement some sort of caching if that was a concern (shouldn't be > > > though). > > > > * For consistency within the whole DagRun, the scheduler absolutely > has > > > to > > > > read the right version. If tasks got removed they would never ge
Re: How to add hooks for strong deployment consistency?
I'm curious to hear which DagFetcher abstraction people would build or want to use. So far it sounded like the most popular and flexible approach would be a `GitDagFetcher` where all SHAs and refs become a possibility, as opposed to say a TarballOnS3DagFetcher which would require more manual artifact management and versioning, which represent additional [human] workflow on top of the already existing git-based workflow. One way I've seen this done before is by using this Git fuse (file system in user space) hack that creates a virtual filesystem where all SHAs and refs in the Git repo are exposed as a subfolder, and under each ref subfolder the whole repo sits as of that ref. Of course all the files are virtual and fetched at access time by the virtual filesystem using the git api. So if you simply point the DagBag loader to the right [virtual] directory, it will import the right version of the DAG. In the git world, the alternative to that is managing temp folders and doing shallow clones which seems like much more of a headache. Note that one tradeoff is that if git and whatever it depends has then a need to be highly available. Max On Wed, Feb 28, 2018 at 6:55 PM, David Capwell wrote: > Thanks for all the details! With a pluggable fetcher we would be able to > add our own logic for how to fetch so sounds like a good place to start for > something like this! > > On Wed, Feb 28, 2018, 4:39 PM Joy Gao wrote: > > > +1 on DagFetcher abstraction, very airflow-esque :) > > > > On Wed, Feb 28, 2018 at 11:25 AM, Maxime Beauchemin > > wrote: > > > Addressing a few of your questions / concerns: > > > > > > * The scheduler uses a multiprocess queue to queue up tasks, each > > > subprocess is in charge of a single DAG "scheduler cycle" which > triggers > > > what it can for active DagRuns. Currently it fills the DagBag from the > > > local file system, looking for a specific module where the master > process > > > last saw that DAG. Fetching the DAG is a metadata operation, DAG > > artifacts > > > shouldn't be too large, we can assume that it takes seconds at most to > > > fetch a DAG, which is ok. We generally assume that the scheduler should > > > fully cycle every minute or so. Version-aware DagFetcher could also > > > implement some sort of caching if that was a concern (shouldn't be > > though). > > > * For consistency within the whole DagRun, the scheduler absolutely has > > to > > > read the right version. If tasks got removed they would never get > > scheduled > > > and consistency cannot be achieved. > > > * TaskInstances get created the first time they are identified as > > runnable > > > by the scheduler and are born with a queued status I believe (from > > memory, > > > haven't read the latest code to confirm). The worker double checks and > > sets > > > it as running as part of a database transaction to avoid double-firing. > > > > > > Max > > > > > > On Wed, Feb 28, 2018 at 7:29 AM, Chris Palmer > > wrote: > > > > > >> I'll preface this with the fact that I'm relatively new to Airflow, > and > > >> haven't played around with a lot of the internals. > > >> > > >> I find the idea of a DagFetcher interesting but would we worry about > > >> slowing down the scheduler significantly? If the scheduler is having > to > > >> "fetch" multiple different DAG versions, be it git refs or artifacts > > from > > >> Artifactory, we are talking about adding significant time to each > > scheduler > > >> run. Also how would the scheduler know which DAGs to fetch from where > if > > >> there aren't local files on disk listing those DAGs? Maybe I'm missing > > >> something in the implementation. > > >> > > >> It seems to me that the fetching of the different versions should be > > >> delegated to the Task (or TaskInstance) itself. That ensures we only > > spend > > >> the time to "fetch" the version that is needed when it is needed. One > > down > > >> side might be that each TaskInstance running for the same version of > the > > >> DAG might end up doing the "fetch" independently (duplicating that > > work). > > >> > > >> I think this could be done by adding some version attribute to the > > DagRun > > >> that gets set at creation, and have the scheduler pass that version to > > the > > >> TaskInstances when they are created. You could even extend this so > that > > you > > >> could have an arbitrary set of "executor_parameters" that get set on a > > >> DagRun and are passed to TaskInstances. Then the specific Executor > class > > >> that is running that TaskInstance could handle the > > "executor_parameters" as > > >> it sees fit. > > >> > > >> One thing I'm not clear on is how and when TaskInstances are created. > > When > > >> the scheduler first sees a specific DagRun do all the TaskInstances > get > > >> created immediately, but only some of them get queued? Or does the > > >> scheduler only create those TaskInstances which can be queued right > now? > > >> > > >> In particular if a DagRun gets created and wh
Re: How to add hooks for strong deployment consistency?
Thanks for all the details! With a pluggable fetcher we would be able to add our own logic for how to fetch so sounds like a good place to start for something like this! On Wed, Feb 28, 2018, 4:39 PM Joy Gao wrote: > +1 on DagFetcher abstraction, very airflow-esque :) > > On Wed, Feb 28, 2018 at 11:25 AM, Maxime Beauchemin > wrote: > > Addressing a few of your questions / concerns: > > > > * The scheduler uses a multiprocess queue to queue up tasks, each > > subprocess is in charge of a single DAG "scheduler cycle" which triggers > > what it can for active DagRuns. Currently it fills the DagBag from the > > local file system, looking for a specific module where the master process > > last saw that DAG. Fetching the DAG is a metadata operation, DAG > artifacts > > shouldn't be too large, we can assume that it takes seconds at most to > > fetch a DAG, which is ok. We generally assume that the scheduler should > > fully cycle every minute or so. Version-aware DagFetcher could also > > implement some sort of caching if that was a concern (shouldn't be > though). > > * For consistency within the whole DagRun, the scheduler absolutely has > to > > read the right version. If tasks got removed they would never get > scheduled > > and consistency cannot be achieved. > > * TaskInstances get created the first time they are identified as > runnable > > by the scheduler and are born with a queued status I believe (from > memory, > > haven't read the latest code to confirm). The worker double checks and > sets > > it as running as part of a database transaction to avoid double-firing. > > > > Max > > > > On Wed, Feb 28, 2018 at 7:29 AM, Chris Palmer > wrote: > > > >> I'll preface this with the fact that I'm relatively new to Airflow, and > >> haven't played around with a lot of the internals. > >> > >> I find the idea of a DagFetcher interesting but would we worry about > >> slowing down the scheduler significantly? If the scheduler is having to > >> "fetch" multiple different DAG versions, be it git refs or artifacts > from > >> Artifactory, we are talking about adding significant time to each > scheduler > >> run. Also how would the scheduler know which DAGs to fetch from where if > >> there aren't local files on disk listing those DAGs? Maybe I'm missing > >> something in the implementation. > >> > >> It seems to me that the fetching of the different versions should be > >> delegated to the Task (or TaskInstance) itself. That ensures we only > spend > >> the time to "fetch" the version that is needed when it is needed. One > down > >> side might be that each TaskInstance running for the same version of the > >> DAG might end up doing the "fetch" independently (duplicating that > work). > >> > >> I think this could be done by adding some version attribute to the > DagRun > >> that gets set at creation, and have the scheduler pass that version to > the > >> TaskInstances when they are created. You could even extend this so that > you > >> could have an arbitrary set of "executor_parameters" that get set on a > >> DagRun and are passed to TaskInstances. Then the specific Executor class > >> that is running that TaskInstance could handle the > "executor_parameters" as > >> it sees fit. > >> > >> One thing I'm not clear on is how and when TaskInstances are created. > When > >> the scheduler first sees a specific DagRun do all the TaskInstances get > >> created immediately, but only some of them get queued? Or does the > >> scheduler only create those TaskInstances which can be queued right now? > >> > >> In particular if a DagRun gets created and while it is running the DAG > is > >> updated and a new Task is added, will the scheduler pick up that new > Task > >> for the running DagRun? If the answer is yes, then my suggestion above > >> would run the risk of scheduling a Task for a DAG version where that > Task > >> didn't exist. I'm sure you could handle that somewhat gracefully but > it's a > >> bit ugly. > >> > >> Chris > >> > >> On Wed, Feb 28, 2018 at 2:05 AM, Maxime Beauchemin < > >> maximebeauche...@gmail.com> wrote: > >> > >> > At a higher level I want to say a few things about the idea of > enforcing > >> > version consistency within a DagRun. > >> > > >> > One thing we've been talking about is the need for a "DagFetcher" > >> > abstraction, where it's first implementation that would replace and > mimic > >> > the current one would be "FileSystemDagFetcher". One specific > DagFetcher > >> > implementation may or may not support version semantics, but if it > does > >> > should be able to receive a version id and return the proper version > of > >> the > >> > DAG object. For instance that first "FileSystemDagFetcher" would not > >> > support version semantic, but perhaps a "GitRepoDagFetcher" would, or > an > >> > "ArtifactoryDagFetcher", or "TarballInS3DagFetcher" may as well. > >> > > >> > Of course that assumes that the scheduler knows and stores the active > >> > version number when generating a new DagR
Re: How to add hooks for strong deployment consistency?
+1 on DagFetcher abstraction, very airflow-esque :) On Wed, Feb 28, 2018 at 11:25 AM, Maxime Beauchemin wrote: > Addressing a few of your questions / concerns: > > * The scheduler uses a multiprocess queue to queue up tasks, each > subprocess is in charge of a single DAG "scheduler cycle" which triggers > what it can for active DagRuns. Currently it fills the DagBag from the > local file system, looking for a specific module where the master process > last saw that DAG. Fetching the DAG is a metadata operation, DAG artifacts > shouldn't be too large, we can assume that it takes seconds at most to > fetch a DAG, which is ok. We generally assume that the scheduler should > fully cycle every minute or so. Version-aware DagFetcher could also > implement some sort of caching if that was a concern (shouldn't be though). > * For consistency within the whole DagRun, the scheduler absolutely has to > read the right version. If tasks got removed they would never get scheduled > and consistency cannot be achieved. > * TaskInstances get created the first time they are identified as runnable > by the scheduler and are born with a queued status I believe (from memory, > haven't read the latest code to confirm). The worker double checks and sets > it as running as part of a database transaction to avoid double-firing. > > Max > > On Wed, Feb 28, 2018 at 7:29 AM, Chris Palmer wrote: > >> I'll preface this with the fact that I'm relatively new to Airflow, and >> haven't played around with a lot of the internals. >> >> I find the idea of a DagFetcher interesting but would we worry about >> slowing down the scheduler significantly? If the scheduler is having to >> "fetch" multiple different DAG versions, be it git refs or artifacts from >> Artifactory, we are talking about adding significant time to each scheduler >> run. Also how would the scheduler know which DAGs to fetch from where if >> there aren't local files on disk listing those DAGs? Maybe I'm missing >> something in the implementation. >> >> It seems to me that the fetching of the different versions should be >> delegated to the Task (or TaskInstance) itself. That ensures we only spend >> the time to "fetch" the version that is needed when it is needed. One down >> side might be that each TaskInstance running for the same version of the >> DAG might end up doing the "fetch" independently (duplicating that work). >> >> I think this could be done by adding some version attribute to the DagRun >> that gets set at creation, and have the scheduler pass that version to the >> TaskInstances when they are created. You could even extend this so that you >> could have an arbitrary set of "executor_parameters" that get set on a >> DagRun and are passed to TaskInstances. Then the specific Executor class >> that is running that TaskInstance could handle the "executor_parameters" as >> it sees fit. >> >> One thing I'm not clear on is how and when TaskInstances are created. When >> the scheduler first sees a specific DagRun do all the TaskInstances get >> created immediately, but only some of them get queued? Or does the >> scheduler only create those TaskInstances which can be queued right now? >> >> In particular if a DagRun gets created and while it is running the DAG is >> updated and a new Task is added, will the scheduler pick up that new Task >> for the running DagRun? If the answer is yes, then my suggestion above >> would run the risk of scheduling a Task for a DAG version where that Task >> didn't exist. I'm sure you could handle that somewhat gracefully but it's a >> bit ugly. >> >> Chris >> >> On Wed, Feb 28, 2018 at 2:05 AM, Maxime Beauchemin < >> maximebeauche...@gmail.com> wrote: >> >> > At a higher level I want to say a few things about the idea of enforcing >> > version consistency within a DagRun. >> > >> > One thing we've been talking about is the need for a "DagFetcher" >> > abstraction, where it's first implementation that would replace and mimic >> > the current one would be "FileSystemDagFetcher". One specific DagFetcher >> > implementation may or may not support version semantics, but if it does >> > should be able to receive a version id and return the proper version of >> the >> > DAG object. For instance that first "FileSystemDagFetcher" would not >> > support version semantic, but perhaps a "GitRepoDagFetcher" would, or an >> > "ArtifactoryDagFetcher", or "TarballInS3DagFetcher" may as well. >> > >> > Of course that assumes that the scheduler knows and stores the active >> > version number when generating a new DagRun, and for that information to >> be >> > leveraged on subsequent scheduler cycles and on workers when task are >> > executed. >> > >> > This could also enable things like "remote" backfills (non local, >> > parallelized) of a DAG definition that's on an arbitrary git ref >> (assuming >> > a "GitRepoDagFetcher"). >> > >> > There are [perhaps] unintuitive implications where clearing a single task >> > would then re-run the old DAG defini
Re: How to add hooks for strong deployment consistency?
Addressing a few of your questions / concerns: * The scheduler uses a multiprocess queue to queue up tasks, each subprocess is in charge of a single DAG "scheduler cycle" which triggers what it can for active DagRuns. Currently it fills the DagBag from the local file system, looking for a specific module where the master process last saw that DAG. Fetching the DAG is a metadata operation, DAG artifacts shouldn't be too large, we can assume that it takes seconds at most to fetch a DAG, which is ok. We generally assume that the scheduler should fully cycle every minute or so. Version-aware DagFetcher could also implement some sort of caching if that was a concern (shouldn't be though). * For consistency within the whole DagRun, the scheduler absolutely has to read the right version. If tasks got removed they would never get scheduled and consistency cannot be achieved. * TaskInstances get created the first time they are identified as runnable by the scheduler and are born with a queued status I believe (from memory, haven't read the latest code to confirm). The worker double checks and sets it as running as part of a database transaction to avoid double-firing. Max On Wed, Feb 28, 2018 at 7:29 AM, Chris Palmer wrote: > I'll preface this with the fact that I'm relatively new to Airflow, and > haven't played around with a lot of the internals. > > I find the idea of a DagFetcher interesting but would we worry about > slowing down the scheduler significantly? If the scheduler is having to > "fetch" multiple different DAG versions, be it git refs or artifacts from > Artifactory, we are talking about adding significant time to each scheduler > run. Also how would the scheduler know which DAGs to fetch from where if > there aren't local files on disk listing those DAGs? Maybe I'm missing > something in the implementation. > > It seems to me that the fetching of the different versions should be > delegated to the Task (or TaskInstance) itself. That ensures we only spend > the time to "fetch" the version that is needed when it is needed. One down > side might be that each TaskInstance running for the same version of the > DAG might end up doing the "fetch" independently (duplicating that work). > > I think this could be done by adding some version attribute to the DagRun > that gets set at creation, and have the scheduler pass that version to the > TaskInstances when they are created. You could even extend this so that you > could have an arbitrary set of "executor_parameters" that get set on a > DagRun and are passed to TaskInstances. Then the specific Executor class > that is running that TaskInstance could handle the "executor_parameters" as > it sees fit. > > One thing I'm not clear on is how and when TaskInstances are created. When > the scheduler first sees a specific DagRun do all the TaskInstances get > created immediately, but only some of them get queued? Or does the > scheduler only create those TaskInstances which can be queued right now? > > In particular if a DagRun gets created and while it is running the DAG is > updated and a new Task is added, will the scheduler pick up that new Task > for the running DagRun? If the answer is yes, then my suggestion above > would run the risk of scheduling a Task for a DAG version where that Task > didn't exist. I'm sure you could handle that somewhat gracefully but it's a > bit ugly. > > Chris > > On Wed, Feb 28, 2018 at 2:05 AM, Maxime Beauchemin < > maximebeauche...@gmail.com> wrote: > > > At a higher level I want to say a few things about the idea of enforcing > > version consistency within a DagRun. > > > > One thing we've been talking about is the need for a "DagFetcher" > > abstraction, where it's first implementation that would replace and mimic > > the current one would be "FileSystemDagFetcher". One specific DagFetcher > > implementation may or may not support version semantics, but if it does > > should be able to receive a version id and return the proper version of > the > > DAG object. For instance that first "FileSystemDagFetcher" would not > > support version semantic, but perhaps a "GitRepoDagFetcher" would, or an > > "ArtifactoryDagFetcher", or "TarballInS3DagFetcher" may as well. > > > > Of course that assumes that the scheduler knows and stores the active > > version number when generating a new DagRun, and for that information to > be > > leveraged on subsequent scheduler cycles and on workers when task are > > executed. > > > > This could also enable things like "remote" backfills (non local, > > parallelized) of a DAG definition that's on an arbitrary git ref > (assuming > > a "GitRepoDagFetcher"). > > > > There are [perhaps] unintuitive implications where clearing a single task > > would then re-run the old DAG definition on that task (since the version > > was stamped in the DagRun and hasn't changed), but deleting/recreating a > > DagRun would run the latest version (or any other version that may be > > specified for that matter). > >
Re: How to add hooks for strong deployment consistency?
I'll preface this with the fact that I'm relatively new to Airflow, and haven't played around with a lot of the internals. I find the idea of a DagFetcher interesting but would we worry about slowing down the scheduler significantly? If the scheduler is having to "fetch" multiple different DAG versions, be it git refs or artifacts from Artifactory, we are talking about adding significant time to each scheduler run. Also how would the scheduler know which DAGs to fetch from where if there aren't local files on disk listing those DAGs? Maybe I'm missing something in the implementation. It seems to me that the fetching of the different versions should be delegated to the Task (or TaskInstance) itself. That ensures we only spend the time to "fetch" the version that is needed when it is needed. One down side might be that each TaskInstance running for the same version of the DAG might end up doing the "fetch" independently (duplicating that work). I think this could be done by adding some version attribute to the DagRun that gets set at creation, and have the scheduler pass that version to the TaskInstances when they are created. You could even extend this so that you could have an arbitrary set of "executor_parameters" that get set on a DagRun and are passed to TaskInstances. Then the specific Executor class that is running that TaskInstance could handle the "executor_parameters" as it sees fit. One thing I'm not clear on is how and when TaskInstances are created. When the scheduler first sees a specific DagRun do all the TaskInstances get created immediately, but only some of them get queued? Or does the scheduler only create those TaskInstances which can be queued right now? In particular if a DagRun gets created and while it is running the DAG is updated and a new Task is added, will the scheduler pick up that new Task for the running DagRun? If the answer is yes, then my suggestion above would run the risk of scheduling a Task for a DAG version where that Task didn't exist. I'm sure you could handle that somewhat gracefully but it's a bit ugly. Chris On Wed, Feb 28, 2018 at 2:05 AM, Maxime Beauchemin < maximebeauche...@gmail.com> wrote: > At a higher level I want to say a few things about the idea of enforcing > version consistency within a DagRun. > > One thing we've been talking about is the need for a "DagFetcher" > abstraction, where it's first implementation that would replace and mimic > the current one would be "FileSystemDagFetcher". One specific DagFetcher > implementation may or may not support version semantics, but if it does > should be able to receive a version id and return the proper version of the > DAG object. For instance that first "FileSystemDagFetcher" would not > support version semantic, but perhaps a "GitRepoDagFetcher" would, or an > "ArtifactoryDagFetcher", or "TarballInS3DagFetcher" may as well. > > Of course that assumes that the scheduler knows and stores the active > version number when generating a new DagRun, and for that information to be > leveraged on subsequent scheduler cycles and on workers when task are > executed. > > This could also enable things like "remote" backfills (non local, > parallelized) of a DAG definition that's on an arbitrary git ref (assuming > a "GitRepoDagFetcher"). > > There are [perhaps] unintuitive implications where clearing a single task > would then re-run the old DAG definition on that task (since the version > was stamped in the DagRun and hasn't changed), but deleting/recreating a > DagRun would run the latest version (or any other version that may be > specified for that matter). > > I'm unclear on how much work that represents exactly, but it's certainly > doable and may only require to change part of the DagBag class and a few > other places. > > Max > > On Tue, Feb 27, 2018 at 6:48 PM, David Capwell wrote: > > > Thanks for your feedback! > > > > Option 1 is a non-starter for us. The reason is we have DAGs that take 9+ > > hours to run. > > > > Option 2 is more where my mind was going, but it's rather large. How I > see > > it you need a MVCC DagBag that's aware of multiple versions (what > provides > > version?). Assuming you can track active dag runs pointing to which > > versions you know how to cleanup (fine with external). The pro here is > you > > have snapshot isolation for dag_run, con is more bookkeeping and require > > deploy to work with this (last part may be a good thing though). > > > > The only other option I can think of is to lock deploy so the system only > > picks up new versions when no dag_run holds the lock. This is flawed for > > many reasons, but breaks horrible for dag_runs that takes minutes (I > assume > > 99% do). > > > > > > > > On Tue, Feb 27, 2018, 4:50 PM Joy Gao wrote: > > > > > Hi David! > > > > > > Thank you for clarifying, I think I understand your concern now. We > > > currently also work around this by making sure a dag is turned off > > > when we deploy a new version. We also make sur
Re: How to add hooks for strong deployment consistency?
At a higher level I want to say a few things about the idea of enforcing version consistency within a DagRun. One thing we've been talking about is the need for a "DagFetcher" abstraction, where it's first implementation that would replace and mimic the current one would be "FileSystemDagFetcher". One specific DagFetcher implementation may or may not support version semantics, but if it does should be able to receive a version id and return the proper version of the DAG object. For instance that first "FileSystemDagFetcher" would not support version semantic, but perhaps a "GitRepoDagFetcher" would, or an "ArtifactoryDagFetcher", or "TarballInS3DagFetcher" may as well. Of course that assumes that the scheduler knows and stores the active version number when generating a new DagRun, and for that information to be leveraged on subsequent scheduler cycles and on workers when task are executed. This could also enable things like "remote" backfills (non local, parallelized) of a DAG definition that's on an arbitrary git ref (assuming a "GitRepoDagFetcher"). There are [perhaps] unintuitive implications where clearing a single task would then re-run the old DAG definition on that task (since the version was stamped in the DagRun and hasn't changed), but deleting/recreating a DagRun would run the latest version (or any other version that may be specified for that matter). I'm unclear on how much work that represents exactly, but it's certainly doable and may only require to change part of the DagBag class and a few other places. Max On Tue, Feb 27, 2018 at 6:48 PM, David Capwell wrote: > Thanks for your feedback! > > Option 1 is a non-starter for us. The reason is we have DAGs that take 9+ > hours to run. > > Option 2 is more where my mind was going, but it's rather large. How I see > it you need a MVCC DagBag that's aware of multiple versions (what provides > version?). Assuming you can track active dag runs pointing to which > versions you know how to cleanup (fine with external). The pro here is you > have snapshot isolation for dag_run, con is more bookkeeping and require > deploy to work with this (last part may be a good thing though). > > The only other option I can think of is to lock deploy so the system only > picks up new versions when no dag_run holds the lock. This is flawed for > many reasons, but breaks horrible for dag_runs that takes minutes (I assume > 99% do). > > > > On Tue, Feb 27, 2018, 4:50 PM Joy Gao wrote: > > > Hi David! > > > > Thank you for clarifying, I think I understand your concern now. We > > currently also work around this by making sure a dag is turned off > > when we deploy a new version. We also make sure our jobs are > > idempotent and retry-enabled in the case when we forget to turn off > > the job, so the issue hasn't caused us too much headache. > > > > I do agree that it would be nice for Airflow to have the option to > > guarantee a single version of dag per dag run. I see two approaches: > > > > (1) If a dag is updated, the current dagrun fails and/or retries. > > (2) If a dag is updated, the current dagrun continues but uses version > > before the update. > > > > (1) requires some mechanism to compare dag generations. One option is > > to hash the dagfile and storing that value to the dagrun table, and > > compare against it each time a task is running. And in the case if the > > hash value is different, update the hash value, then fail/retry the > > dag. I think this is a fairly safe approach. > > > > (2) is trickier. A dag only has a property "fileloc" which tracks the > > location of the dag file, but the actual content of the dag file is > > never versioned. When a task instance starts running, it dynamically > > re-processes the dag file specified by the fileloc, generate all the > > task objects from the dag file, and fetch the task object by task_id > > in order to execute it. So in order to guarantee each dagrun to run a > > specific version, previous versions must be maintained on disk somehow > > (maintaining this information in memory is difficult, since if the > > scheduler/worker shuts down, that information is lost). This makes it > > a pretty big change, and I haven't thought much on how to implement > > it. > > > > I'm personally leaning towards (1) for sake of simplicity. Note that > > some users may not want dag to fail/retry even when dag is updated, so > > this should be an optional feature, not required. > > > > My scheduler-foo isn't that great, so curious what others have to say > > about this. > > > > On Fri, Feb 23, 2018 at 3:12 PM, David Capwell > wrote: > > > Thanks for the reply Joy, let me walk you though things as they are > today > > > > > > 1) we don't stop airflow or disable DAGs while deploying updates to > > logic, > > > this is done live once its released > > > 2) the python script in the DAG folder doesn't actually have DAGs in it > > but > > > is a shim layer to allow us to deploy in a atomic way for a single host > >
Re: How to add hooks for strong deployment consistency?
Thanks for your feedback! Option 1 is a non-starter for us. The reason is we have DAGs that take 9+ hours to run. Option 2 is more where my mind was going, but it's rather large. How I see it you need a MVCC DagBag that's aware of multiple versions (what provides version?). Assuming you can track active dag runs pointing to which versions you know how to cleanup (fine with external). The pro here is you have snapshot isolation for dag_run, con is more bookkeeping and require deploy to work with this (last part may be a good thing though). The only other option I can think of is to lock deploy so the system only picks up new versions when no dag_run holds the lock. This is flawed for many reasons, but breaks horrible for dag_runs that takes minutes (I assume 99% do). On Tue, Feb 27, 2018, 4:50 PM Joy Gao wrote: > Hi David! > > Thank you for clarifying, I think I understand your concern now. We > currently also work around this by making sure a dag is turned off > when we deploy a new version. We also make sure our jobs are > idempotent and retry-enabled in the case when we forget to turn off > the job, so the issue hasn't caused us too much headache. > > I do agree that it would be nice for Airflow to have the option to > guarantee a single version of dag per dag run. I see two approaches: > > (1) If a dag is updated, the current dagrun fails and/or retries. > (2) If a dag is updated, the current dagrun continues but uses version > before the update. > > (1) requires some mechanism to compare dag generations. One option is > to hash the dagfile and storing that value to the dagrun table, and > compare against it each time a task is running. And in the case if the > hash value is different, update the hash value, then fail/retry the > dag. I think this is a fairly safe approach. > > (2) is trickier. A dag only has a property "fileloc" which tracks the > location of the dag file, but the actual content of the dag file is > never versioned. When a task instance starts running, it dynamically > re-processes the dag file specified by the fileloc, generate all the > task objects from the dag file, and fetch the task object by task_id > in order to execute it. So in order to guarantee each dagrun to run a > specific version, previous versions must be maintained on disk somehow > (maintaining this information in memory is difficult, since if the > scheduler/worker shuts down, that information is lost). This makes it > a pretty big change, and I haven't thought much on how to implement > it. > > I'm personally leaning towards (1) for sake of simplicity. Note that > some users may not want dag to fail/retry even when dag is updated, so > this should be an optional feature, not required. > > My scheduler-foo isn't that great, so curious what others have to say > about this. > > On Fri, Feb 23, 2018 at 3:12 PM, David Capwell wrote: > > Thanks for the reply Joy, let me walk you though things as they are today > > > > 1) we don't stop airflow or disable DAGs while deploying updates to > logic, > > this is done live once its released > > 2) the python script in the DAG folder doesn't actually have DAGs in it > but > > is a shim layer to allow us to deploy in a atomic way for a single host > > 2.1) this script reads a file on local disk (less than disk page size) > to > > find latest git commit deployed > > 2.2) re-does the airflow DAG load process but pointing to the git > commit > > path > > > > Example directory structure > > > > /airflow/dags/shim.py > > /airflow/real_dags/ > > /latest # pointer to latest commit > > /[git commit]/ > > > > This is how we make sure deploys are consistent within a single task. > > > > > > Now, lets assume we have a fully atomic commit process and are able to > > upgrade DAGs at the exact same moment. > > > > At time T0 the scheduler knows of DAG V1 and schedules two tasks, Task1, > > and Task2 > > At time T1 Task1 is picked up by Worker1, so starts executing the task > (V1 > > logic) > > At time T2 deploy commit happens, current DAG version: V2 > > At time T3, Task2 is picked up by Worker2, so starts executing the task > (V2 > > logic) > > > > In many cases this isn't really a problem (tuning config change to hadoop > > job), but as we have more people using Airflow this is causing a lot of > > time spent debugging why production acted differently than expected (the > > problem was already fixed... why is it still here?). We also see that > some > > tasks expect a given behavior from other tasks, and since they live in > the > > same git repo they can modify both tasks at the same time if a breaking > > change is needed, but when this rolls out to prod there isn't a way to do > > this other than turn off the DAG, and login to all hosts to verify fully > > deployed. > > > > We would like to remove this confusion and make generations/versions > (same > > thing really) exposed to users and make sure for a single dag_ru
Re: How to add hooks for strong deployment consistency?
Hi David! Thank you for clarifying, I think I understand your concern now. We currently also work around this by making sure a dag is turned off when we deploy a new version. We also make sure our jobs are idempotent and retry-enabled in the case when we forget to turn off the job, so the issue hasn't caused us too much headache. I do agree that it would be nice for Airflow to have the option to guarantee a single version of dag per dag run. I see two approaches: (1) If a dag is updated, the current dagrun fails and/or retries. (2) If a dag is updated, the current dagrun continues but uses version before the update. (1) requires some mechanism to compare dag generations. One option is to hash the dagfile and storing that value to the dagrun table, and compare against it each time a task is running. And in the case if the hash value is different, update the hash value, then fail/retry the dag. I think this is a fairly safe approach. (2) is trickier. A dag only has a property "fileloc" which tracks the location of the dag file, but the actual content of the dag file is never versioned. When a task instance starts running, it dynamically re-processes the dag file specified by the fileloc, generate all the task objects from the dag file, and fetch the task object by task_id in order to execute it. So in order to guarantee each dagrun to run a specific version, previous versions must be maintained on disk somehow (maintaining this information in memory is difficult, since if the scheduler/worker shuts down, that information is lost). This makes it a pretty big change, and I haven't thought much on how to implement it. I'm personally leaning towards (1) for sake of simplicity. Note that some users may not want dag to fail/retry even when dag is updated, so this should be an optional feature, not required. My scheduler-foo isn't that great, so curious what others have to say about this. On Fri, Feb 23, 2018 at 3:12 PM, David Capwell wrote: > Thanks for the reply Joy, let me walk you though things as they are today > > 1) we don't stop airflow or disable DAGs while deploying updates to logic, > this is done live once its released > 2) the python script in the DAG folder doesn't actually have DAGs in it but > is a shim layer to allow us to deploy in a atomic way for a single host > 2.1) this script reads a file on local disk (less than disk page size) to > find latest git commit deployed > 2.2) re-does the airflow DAG load process but pointing to the git commit > path > > Example directory structure > > /airflow/dags/shim.py > /airflow/real_dags/ > /latest # pointer to latest commit > /[git commit]/ > > This is how we make sure deploys are consistent within a single task. > > > Now, lets assume we have a fully atomic commit process and are able to > upgrade DAGs at the exact same moment. > > At time T0 the scheduler knows of DAG V1 and schedules two tasks, Task1, > and Task2 > At time T1 Task1 is picked up by Worker1, so starts executing the task (V1 > logic) > At time T2 deploy commit happens, current DAG version: V2 > At time T3, Task2 is picked up by Worker2, so starts executing the task (V2 > logic) > > In many cases this isn't really a problem (tuning config change to hadoop > job), but as we have more people using Airflow this is causing a lot of > time spent debugging why production acted differently than expected (the > problem was already fixed... why is it still here?). We also see that some > tasks expect a given behavior from other tasks, and since they live in the > same git repo they can modify both tasks at the same time if a breaking > change is needed, but when this rolls out to prod there isn't a way to do > this other than turn off the DAG, and login to all hosts to verify fully > deployed. > > We would like to remove this confusion and make generations/versions (same > thing really) exposed to users and make sure for a single dag_run only one > version is used. > > I hope this is more clear. > > On Fri, Feb 23, 2018 at 1:37 PM, Joy Gao wrote: > >> Hi David, >> >> Do you mind providing a concrete example of the scenario in which >> scheduler/workers see different states (I'm not 100% sure if I understood >> the issue at hand). >> >> And by same dag generation, are you referring to the dag version? (DAG >> version is currently not supported at all, but I can see it being a >> building block for future use cases). >> >> Joy >> >> On Fri, Feb 23, 2018 at 1:00 PM, David Capwell wrote: >> >> > My current thinking is to add a field to the dag table that is optional >> and >> > provided by the dag. We currently intercept the load path do could use >> this >> > field to make sure we load the same generation. My concern here is the >> > interaction with the scheduler, not as familiar with that logic to >> predict >> > corner cases were this would fail. >> > >> > Any other recommendations for how this could be done? >> > >> > On Mon,
Re: How to add hooks for strong deployment consistency?
Thanks for the reply Joy, let me walk you though things as they are today 1) we don't stop airflow or disable DAGs while deploying updates to logic, this is done live once its released 2) the python script in the DAG folder doesn't actually have DAGs in it but is a shim layer to allow us to deploy in a atomic way for a single host 2.1) this script reads a file on local disk (less than disk page size) to find latest git commit deployed 2.2) re-does the airflow DAG load process but pointing to the git commit path Example directory structure /airflow/dags/shim.py /airflow/real_dags/ /latest # pointer to latest commit /[git commit]/ This is how we make sure deploys are consistent within a single task. Now, lets assume we have a fully atomic commit process and are able to upgrade DAGs at the exact same moment. At time T0 the scheduler knows of DAG V1 and schedules two tasks, Task1, and Task2 At time T1 Task1 is picked up by Worker1, so starts executing the task (V1 logic) At time T2 deploy commit happens, current DAG version: V2 At time T3, Task2 is picked up by Worker2, so starts executing the task (V2 logic) In many cases this isn't really a problem (tuning config change to hadoop job), but as we have more people using Airflow this is causing a lot of time spent debugging why production acted differently than expected (the problem was already fixed... why is it still here?). We also see that some tasks expect a given behavior from other tasks, and since they live in the same git repo they can modify both tasks at the same time if a breaking change is needed, but when this rolls out to prod there isn't a way to do this other than turn off the DAG, and login to all hosts to verify fully deployed. We would like to remove this confusion and make generations/versions (same thing really) exposed to users and make sure for a single dag_run only one version is used. I hope this is more clear. On Fri, Feb 23, 2018 at 1:37 PM, Joy Gao wrote: > Hi David, > > Do you mind providing a concrete example of the scenario in which > scheduler/workers see different states (I'm not 100% sure if I understood > the issue at hand). > > And by same dag generation, are you referring to the dag version? (DAG > version is currently not supported at all, but I can see it being a > building block for future use cases). > > Joy > > On Fri, Feb 23, 2018 at 1:00 PM, David Capwell wrote: > > > My current thinking is to add a field to the dag table that is optional > and > > provided by the dag. We currently intercept the load path do could use > this > > field to make sure we load the same generation. My concern here is the > > interaction with the scheduler, not as familiar with that logic to > predict > > corner cases were this would fail. > > > > Any other recommendations for how this could be done? > > > > On Mon, Feb 19, 2018, 10:33 PM David Capwell wrote: > > > > > We have been using airflow for logic that delegates to other systems so > > > inject a task all tasks depends to make sure all resources used are the > > > same for all tasks in the dag. This works well for tasks that delegates > > to > > > external systems but people are starting to need to run logic in > airflow > > > and the fact that scheduler and all workers can see different states is > > > causing issues > > > > > > We can make sure that all the code is deployed in a consistent way but > > > need help from the scheduler to tell the workers the current generation > > for > > > a DAG. > > > > > > My question is, what would be the best way to modify airflow to allow > > DAGs > > > to define a generation value that the scheduler could send to workers? > > > > > > Thanks > > > > > >
Re: How to add hooks for strong deployment consistency?
Hi David, Do you mind providing a concrete example of the scenario in which scheduler/workers see different states (I'm not 100% sure if I understood the issue at hand). And by same dag generation, are you referring to the dag version? (DAG version is currently not supported at all, but I can see it being a building block for future use cases). Joy On Fri, Feb 23, 2018 at 1:00 PM, David Capwell wrote: > My current thinking is to add a field to the dag table that is optional and > provided by the dag. We currently intercept the load path do could use this > field to make sure we load the same generation. My concern here is the > interaction with the scheduler, not as familiar with that logic to predict > corner cases were this would fail. > > Any other recommendations for how this could be done? > > On Mon, Feb 19, 2018, 10:33 PM David Capwell wrote: > > > We have been using airflow for logic that delegates to other systems so > > inject a task all tasks depends to make sure all resources used are the > > same for all tasks in the dag. This works well for tasks that delegates > to > > external systems but people are starting to need to run logic in airflow > > and the fact that scheduler and all workers can see different states is > > causing issues > > > > We can make sure that all the code is deployed in a consistent way but > > need help from the scheduler to tell the workers the current generation > for > > a DAG. > > > > My question is, what would be the best way to modify airflow to allow > DAGs > > to define a generation value that the scheduler could send to workers? > > > > Thanks > > >
Re: How to add hooks for strong deployment consistency?
My current thinking is to add a field to the dag table that is optional and provided by the dag. We currently intercept the load path do could use this field to make sure we load the same generation. My concern here is the interaction with the scheduler, not as familiar with that logic to predict corner cases were this would fail. Any other recommendations for how this could be done? On Mon, Feb 19, 2018, 10:33 PM David Capwell wrote: > We have been using airflow for logic that delegates to other systems so > inject a task all tasks depends to make sure all resources used are the > same for all tasks in the dag. This works well for tasks that delegates to > external systems but people are starting to need to run logic in airflow > and the fact that scheduler and all workers can see different states is > causing issues > > We can make sure that all the code is deployed in a consistent way but > need help from the scheduler to tell the workers the current generation for > a DAG. > > My question is, what would be the best way to modify airflow to allow DAGs > to define a generation value that the scheduler could send to workers? > > Thanks >
How to add hooks for strong deployment consistency?
We have been using airflow for logic that delegates to other systems so inject a task all tasks depends to make sure all resources used are the same for all tasks in the dag. This works well for tasks that delegates to external systems but people are starting to need to run logic in airflow and the fact that scheduler and all workers can see different states is causing issues We can make sure that all the code is deployed in a consistent way but need help from the scheduler to tell the workers the current generation for a DAG. My question is, what would be the best way to modify airflow to allow DAGs to define a generation value that the scheduler could send to workers? Thanks
