You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@airflow.apache.org by Georg Walther <ge...@markovian.com> on 2016/12/20 14:48:59 UTC
Re: Dynamic creation of DAG
Hi Maxime,
thanks for the link you posted.
The example provided in the documentation on dynamically adding DAGs does
not work for us - even bearing in mind your addendum you pointed to.
Our setup is as follows:
We deploy the webserver, scheduler and four workers (using the celery
executor) in one separate docker container each (all containers are
connected via the same Docker network).
The dag bag directory pointed to by `dags_folder` in airflow.cfg contains
one Python module that implements a DAG factory:
The DAG factory fetches DAG parameters from a database and instantiates the
appropriate DAG object (complete with all tasks).
Doing `globals()[dag_id] = dag_object` in the DAG factory code has no
effect i.e. the `dag_object` does not appear in our webserver and is not
executed.
The DAG factory code is executed in one of the four aforementioned workers.
I have attempted some DAG factory code that adds the DAG object to the
airflow database, however all I end up with are the corresponding `dag_id`s
listed in the webserver as "unavailable":
(... DAG factory class code ...)
@provide_session
def _add_dag_to_airflow(self, dag, session=None): # `dag` is the DAG
object instantiated by the DAG factory
for attribute, default in [('is_subdag', False)]:
dag = self._add_dag_attribute(dag, attribute, default)
DAG.sync_to_db(dag, dag.owner, datetime.utcnow())
dag_pickle = dag.pickle()
dag_model = session.\
query(DagModel).\
filter(DagModel.dag_id == dag.dag_id).\
first()
dag_model.pickle_id = dag_pickle.id
dag_model.is_paused = False
session.merge(dag_model)
session.commit()
bag = DagBag('', include_examples=False)
bag.bag_dag(dag, None, None)
queue = []
job = SchedulerJob(dag_id=dag.dag_id)
job._process_dags(bag, [dag], queue)
for dag_id, task_id, execution_date in queue:
task = dag.get_task(task_id)
task_instance = TaskInstance(task, execution_date)
task_instance.state = State.SCHEDULED
session.merge(task_instance)
session.commit()
Any help on this workflow would be greatly appreciated!
Best,
Georg
On Tue, Nov 22, 2016 at 10:46 PM, Maxime Beauchemin <
maximebeauchemin@gmail.com> wrote:
> I added a paragraph to the FAQ entry: "How can I create DAGs dynamically?"
>
> The GH link above points straight to that paragraph. It's related to an
> issue that was discussed on this mailing list last week.
>
> I agree that working examples are ideal, and they can help with test
> coverage as well.
>
> Max
>
> On Tue, Nov 22, 2016 at 1:10 PM, siddharth anand <sa...@apache.org>
> wrote:
>
> > Hi Max,
> > Which part in the above PR is related to dynamic dags?
> >
> > When thinking about adding documentation about functionality, I propose
> the
> > community bias towards adding working examples and test coverage. We
> offer
> > a quick start (which by the way needs some updates - for example, why
> does
> > it not start airflow-scheduler after starting the webserver?), but then
> > folks get stuck in how to write DAGs and use the full range of Airflow
> > capabilities. This is where examples and better test coverage help keep
> > newbies productive.
> >
> > Perhaps the examples and tests can be upgraded to show a fuller set of
> > dynamic dag capabilities?
> >
> > -s
> >
> > On Mon, Nov 21, 2016 at 7:55 PM, Maxime Beauchemin <
> > maximebeauchemin@gmail.com> wrote:
> >
> > > I just added a bit of information about dynamic DAG creation here:
> > > https://github.com/apache/incubator-airflow/pull/1889/files#diff-
> > > c6f0a0722c6a2f86277535d7bcec7f8cR162
> > >
> > > Let me know if it helps.
> > >
> > > Max
> > >
> > > On Mon, Nov 21, 2016 at 2:58 AM, Deepak Kumar Malladi <
> > > kapeed2091@gmail.com>
> > > wrote:
> > >
> > > > Hi,
> > > >
> > > > I want to dynamically create DAG during run time. I tried the snippet
> > > given
> > > > in the documentation. But it didnt work for me.
> > > >
> > > > Any pointer on how to trigger DAGs which aren't actually present in
> DAG
> > > > folder but are created through code execution (dynamically created)?
> > > >
> > > >
> > > > Thanks & Regards,
> > > > Deepak
> > > >
> > >
> >
>
Re: Dynamic creation of DAG
Posted by Georg Walther <ge...@markovian.com>.
Hi Maxime,
to elaborate on our deployment, here's a sketch of our DAG factory and
factory scheduler.
We compile a Docker image with all requisite libraries.
Based on this image we deploy our code in Docker containers.
One deployment configuration we tried uses the LocalExecutor and:
- one Docker container running the webserver
- one Docker container running the scheduler
Another one we tried uses the CeleryExecutor and:
- one Docker container running the webserver
- one Docker container running the scheduler
- four Docker containers each running a worker
In either configuration I do not comprehend how your globals namespace
approach allows the webserver
to see what the scheduler has added to its own global namespace.
If I understand correctly, in the code sample below the scheduler will
execute `factory.run()` thus adding
the generated DAG object to its own global namespace.
How does the webserver pick up the DAG object held in the global namespace
of the scheduler?
And how is the DAG object persisted across scheduler restarts (we restart
our scheduler container periodically as is suggested somewhere in the
documentation).
This is the file tree of our DAGS_FOLDER:
DAGS_FOLDER
|--- factory_scheduler.py
|--- factory_module.py
# factory_scheduler.py
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from factory_module import Factory
dag = DAG(
dag_id='factory_scheduler',
schedule_interval='* * * * *',
default_args={
'owner': 'markovian',
'start_date': datetime(2016, 12, 18),
'retries': 0
}
)
def callable_():
factory = Factory()
factory.run()
task = PythonOperator(
task_id='some_task_id',
python_callable=callable_,
dag=dag
)
# factory_module.py
from airflow import DAG
class Factory:
def run(self):
dag_id, dag = self.get_dag() # get_dag instantiates the DAG
object `dag`
globals()[dag_id] = dag
Thanks,
Georg
On Wed, Dec 21, 2016 at 12:14 AM, Georg Walther <georg.walther@markovian.com
> wrote:
> Hi Maxime,
>
>
> thank you for your response.
> You wrote:
>
> "The
> scheduler (and web server) will walk through your DAGS_FOLDER and find the
> DAG objects in your modules, and schedule them based on their
> configuration."
>
> The DAG objects we need to schedule are created at runtime. They do not
> exist in any module in the DAGS_FOLDER.
>
> Further, we run our airflow deployment with the celery executor so I
> reckon the `globals` namespace of the celery worker where the respective
> DAG object is created at runtime differs from the `globals` namespace of
> the scheduler (worker and scheduler run in separate Docker containers).
>
>
> Best,
>
> Georg
>
>
> On Tue, Dec 20, 2016 at 9:24 PM, Maxime Beauchemin <
> maximebeauchemin@gmail.com> wrote:
>
>> From looking at the code it doesn't seem like you're going in the right
>> direction. All you need to do is create DAG objects and stamp them in the
>> global namespace. This is configuration as code, you shouldn't be using
>> `session` or calling any of the internal things like `bag_dag`. The
>> scheduler (and web server) will walk through your DAGS_FOLDER and find the
>> DAG objects in your modules, and schedule them based on their
>> configuration.
>>
>> Max
>>
>> On Tue, Dec 20, 2016 at 6:48 AM, Georg Walther <
>> georg.walther@markovian.com>
>> wrote:
>>
>> > Hi Maxime,
>> >
>> >
>> > thanks for the link you posted.
>> >
>> > The example provided in the documentation on dynamically adding DAGs
>> does
>> > not work for us - even bearing in mind your addendum you pointed to.
>> >
>> > Our setup is as follows:
>> >
>> > We deploy the webserver, scheduler and four workers (using the celery
>> > executor) in one separate docker container each (all containers are
>> > connected via the same Docker network).
>> >
>> > The dag bag directory pointed to by `dags_folder` in airflow.cfg
>> contains
>> > one Python module that implements a DAG factory:
>> >
>> > The DAG factory fetches DAG parameters from a database and instantiates
>> the
>> > appropriate DAG object (complete with all tasks).
>> > Doing `globals()[dag_id] = dag_object` in the DAG factory code has no
>> > effect i.e. the `dag_object` does not appear in our webserver and is not
>> > executed.
>> >
>> > The DAG factory code is executed in one of the four aforementioned
>> workers.
>> >
>> > I have attempted some DAG factory code that adds the DAG object to the
>> > airflow database, however all I end up with are the corresponding
>> `dag_id`s
>> > listed in the webserver as "unavailable":
>> >
>> > (... DAG factory class code ...)
>> > @provide_session
>> > def _add_dag_to_airflow(self, dag, session=None): # `dag` is the
>> DAG
>> > object instantiated by the DAG factory
>> > for attribute, default in [('is_subdag', False)]:
>> > dag = self._add_dag_attribute(dag, attribute, default)
>> >
>> > DAG.sync_to_db(dag, dag.owner, datetime.utcnow())
>> > dag_pickle = dag.pickle()
>> >
>> > dag_model = session.\
>> > query(DagModel).\
>> > filter(DagModel.dag_id == dag.dag_id).\
>> > first()
>> > dag_model.pickle_id = dag_pickle.id
>> > dag_model.is_paused = False
>> >
>> > session.merge(dag_model)
>> > session.commit()
>> >
>> > bag = DagBag('', include_examples=False)
>> > bag.bag_dag(dag, None, None)
>> >
>> > queue = []
>> > job = SchedulerJob(dag_id=dag.dag_id)
>> > job._process_dags(bag, [dag], queue)
>> >
>> > for dag_id, task_id, execution_date in queue:
>> > task = dag.get_task(task_id)
>> >
>> > task_instance = TaskInstance(task, execution_date)
>> > task_instance.state = State.SCHEDULED
>> >
>> > session.merge(task_instance)
>> > session.commit()
>> >
>> >
>> >
>> > Any help on this workflow would be greatly appreciated!
>> >
>> >
>> > Best,
>> >
>> > Georg
>> >
>> >
>> > On Tue, Nov 22, 2016 at 10:46 PM, Maxime Beauchemin <
>> > maximebeauchemin@gmail.com> wrote:
>> >
>> > > I added a paragraph to the FAQ entry: "How can I create DAGs
>> > dynamically?"
>> > >
>> > > The GH link above points straight to that paragraph. It's related to
>> an
>> > > issue that was discussed on this mailing list last week.
>> > >
>> > > I agree that working examples are ideal, and they can help with test
>> > > coverage as well.
>> > >
>> > > Max
>> > >
>> > > On Tue, Nov 22, 2016 at 1:10 PM, siddharth anand <sa...@apache.org>
>> > > wrote:
>> > >
>> > > > Hi Max,
>> > > > Which part in the above PR is related to dynamic dags?
>> > > >
>> > > > When thinking about adding documentation about functionality, I
>> propose
>> > > the
>> > > > community bias towards adding working examples and test coverage. We
>> > > offer
>> > > > a quick start (which by the way needs some updates - for example,
>> why
>> > > does
>> > > > it not start airflow-scheduler after starting the webserver?), but
>> then
>> > > > folks get stuck in how to write DAGs and use the full range of
>> Airflow
>> > > > capabilities. This is where examples and better test coverage help
>> keep
>> > > > newbies productive.
>> > > >
>> > > > Perhaps the examples and tests can be upgraded to show a fuller set
>> of
>> > > > dynamic dag capabilities?
>> > > >
>> > > > -s
>> > > >
>> > > > On Mon, Nov 21, 2016 at 7:55 PM, Maxime Beauchemin <
>> > > > maximebeauchemin@gmail.com> wrote:
>> > > >
>> > > > > I just added a bit of information about dynamic DAG creation here:
>> > > > > https://github.com/apache/incubator-airflow/pull/1889/files#diff-
>> > > > > c6f0a0722c6a2f86277535d7bcec7f8cR162
>> > > > >
>> > > > > Let me know if it helps.
>> > > > >
>> > > > > Max
>> > > > >
>> > > > > On Mon, Nov 21, 2016 at 2:58 AM, Deepak Kumar Malladi <
>> > > > > kapeed2091@gmail.com>
>> > > > > wrote:
>> > > > >
>> > > > > > Hi,
>> > > > > >
>> > > > > > I want to dynamically create DAG during run time. I tried the
>> > snippet
>> > > > > given
>> > > > > > in the documentation. But it didnt work for me.
>> > > > > >
>> > > > > > Any pointer on how to trigger DAGs which aren't actually
>> present in
>> > > DAG
>> > > > > > folder but are created through code execution (dynamically
>> > created)?
>> > > > > >
>> > > > > >
>> > > > > > Thanks & Regards,
>> > > > > > Deepak
>> > > > > >
>> > > > >
>> > > >
>> > >
>> >
>>
>
>
Re: Dynamic creation of DAG
Posted by Georg Walther <ge...@markovian.com>.
Hi Maxime,
thank you for your response.
You wrote:
"The
scheduler (and web server) will walk through your DAGS_FOLDER and find the
DAG objects in your modules, and schedule them based on their
configuration."
The DAG objects we need to schedule are created at runtime. They do not
exist in any module in the DAGS_FOLDER.
Further, we run our airflow deployment with the celery executor so I reckon
the `globals` namespace of the celery worker where the respective DAG
object is created at runtime differs from the `globals` namespace of the
scheduler (worker and scheduler run in separate Docker containers).
Best,
Georg
On Tue, Dec 20, 2016 at 9:24 PM, Maxime Beauchemin <
maximebeauchemin@gmail.com> wrote:
> From looking at the code it doesn't seem like you're going in the right
> direction. All you need to do is create DAG objects and stamp them in the
> global namespace. This is configuration as code, you shouldn't be using
> `session` or calling any of the internal things like `bag_dag`. The
> scheduler (and web server) will walk through your DAGS_FOLDER and find the
> DAG objects in your modules, and schedule them based on their
> configuration.
>
> Max
>
> On Tue, Dec 20, 2016 at 6:48 AM, Georg Walther <
> georg.walther@markovian.com>
> wrote:
>
> > Hi Maxime,
> >
> >
> > thanks for the link you posted.
> >
> > The example provided in the documentation on dynamically adding DAGs does
> > not work for us - even bearing in mind your addendum you pointed to.
> >
> > Our setup is as follows:
> >
> > We deploy the webserver, scheduler and four workers (using the celery
> > executor) in one separate docker container each (all containers are
> > connected via the same Docker network).
> >
> > The dag bag directory pointed to by `dags_folder` in airflow.cfg contains
> > one Python module that implements a DAG factory:
> >
> > The DAG factory fetches DAG parameters from a database and instantiates
> the
> > appropriate DAG object (complete with all tasks).
> > Doing `globals()[dag_id] = dag_object` in the DAG factory code has no
> > effect i.e. the `dag_object` does not appear in our webserver and is not
> > executed.
> >
> > The DAG factory code is executed in one of the four aforementioned
> workers.
> >
> > I have attempted some DAG factory code that adds the DAG object to the
> > airflow database, however all I end up with are the corresponding
> `dag_id`s
> > listed in the webserver as "unavailable":
> >
> > (... DAG factory class code ...)
> > @provide_session
> > def _add_dag_to_airflow(self, dag, session=None): # `dag` is the DAG
> > object instantiated by the DAG factory
> > for attribute, default in [('is_subdag', False)]:
> > dag = self._add_dag_attribute(dag, attribute, default)
> >
> > DAG.sync_to_db(dag, dag.owner, datetime.utcnow())
> > dag_pickle = dag.pickle()
> >
> > dag_model = session.\
> > query(DagModel).\
> > filter(DagModel.dag_id == dag.dag_id).\
> > first()
> > dag_model.pickle_id = dag_pickle.id
> > dag_model.is_paused = False
> >
> > session.merge(dag_model)
> > session.commit()
> >
> > bag = DagBag('', include_examples=False)
> > bag.bag_dag(dag, None, None)
> >
> > queue = []
> > job = SchedulerJob(dag_id=dag.dag_id)
> > job._process_dags(bag, [dag], queue)
> >
> > for dag_id, task_id, execution_date in queue:
> > task = dag.get_task(task_id)
> >
> > task_instance = TaskInstance(task, execution_date)
> > task_instance.state = State.SCHEDULED
> >
> > session.merge(task_instance)
> > session.commit()
> >
> >
> >
> > Any help on this workflow would be greatly appreciated!
> >
> >
> > Best,
> >
> > Georg
> >
> >
> > On Tue, Nov 22, 2016 at 10:46 PM, Maxime Beauchemin <
> > maximebeauchemin@gmail.com> wrote:
> >
> > > I added a paragraph to the FAQ entry: "How can I create DAGs
> > dynamically?"
> > >
> > > The GH link above points straight to that paragraph. It's related to an
> > > issue that was discussed on this mailing list last week.
> > >
> > > I agree that working examples are ideal, and they can help with test
> > > coverage as well.
> > >
> > > Max
> > >
> > > On Tue, Nov 22, 2016 at 1:10 PM, siddharth anand <sa...@apache.org>
> > > wrote:
> > >
> > > > Hi Max,
> > > > Which part in the above PR is related to dynamic dags?
> > > >
> > > > When thinking about adding documentation about functionality, I
> propose
> > > the
> > > > community bias towards adding working examples and test coverage. We
> > > offer
> > > > a quick start (which by the way needs some updates - for example, why
> > > does
> > > > it not start airflow-scheduler after starting the webserver?), but
> then
> > > > folks get stuck in how to write DAGs and use the full range of
> Airflow
> > > > capabilities. This is where examples and better test coverage help
> keep
> > > > newbies productive.
> > > >
> > > > Perhaps the examples and tests can be upgraded to show a fuller set
> of
> > > > dynamic dag capabilities?
> > > >
> > > > -s
> > > >
> > > > On Mon, Nov 21, 2016 at 7:55 PM, Maxime Beauchemin <
> > > > maximebeauchemin@gmail.com> wrote:
> > > >
> > > > > I just added a bit of information about dynamic DAG creation here:
> > > > > https://github.com/apache/incubator-airflow/pull/1889/files#diff-
> > > > > c6f0a0722c6a2f86277535d7bcec7f8cR162
> > > > >
> > > > > Let me know if it helps.
> > > > >
> > > > > Max
> > > > >
> > > > > On Mon, Nov 21, 2016 at 2:58 AM, Deepak Kumar Malladi <
> > > > > kapeed2091@gmail.com>
> > > > > wrote:
> > > > >
> > > > > > Hi,
> > > > > >
> > > > > > I want to dynamically create DAG during run time. I tried the
> > snippet
> > > > > given
> > > > > > in the documentation. But it didnt work for me.
> > > > > >
> > > > > > Any pointer on how to trigger DAGs which aren't actually present
> in
> > > DAG
> > > > > > folder but are created through code execution (dynamically
> > created)?
> > > > > >
> > > > > >
> > > > > > Thanks & Regards,
> > > > > > Deepak
> > > > > >
> > > > >
> > > >
> > >
> >
>
Re: Dynamic creation of DAG
Posted by Maxime Beauchemin <ma...@gmail.com>.
From looking at the code it doesn't seem like you're going in the right
direction. All you need to do is create DAG objects and stamp them in the
global namespace. This is configuration as code, you shouldn't be using
`session` or calling any of the internal things like `bag_dag`. The
scheduler (and web server) will walk through your DAGS_FOLDER and find the
DAG objects in your modules, and schedule them based on their configuration.
Max
On Tue, Dec 20, 2016 at 6:48 AM, Georg Walther <ge...@markovian.com>
wrote:
> Hi Maxime,
>
>
> thanks for the link you posted.
>
> The example provided in the documentation on dynamically adding DAGs does
> not work for us - even bearing in mind your addendum you pointed to.
>
> Our setup is as follows:
>
> We deploy the webserver, scheduler and four workers (using the celery
> executor) in one separate docker container each (all containers are
> connected via the same Docker network).
>
> The dag bag directory pointed to by `dags_folder` in airflow.cfg contains
> one Python module that implements a DAG factory:
>
> The DAG factory fetches DAG parameters from a database and instantiates the
> appropriate DAG object (complete with all tasks).
> Doing `globals()[dag_id] = dag_object` in the DAG factory code has no
> effect i.e. the `dag_object` does not appear in our webserver and is not
> executed.
>
> The DAG factory code is executed in one of the four aforementioned workers.
>
> I have attempted some DAG factory code that adds the DAG object to the
> airflow database, however all I end up with are the corresponding `dag_id`s
> listed in the webserver as "unavailable":
>
> (... DAG factory class code ...)
> @provide_session
> def _add_dag_to_airflow(self, dag, session=None): # `dag` is the DAG
> object instantiated by the DAG factory
> for attribute, default in [('is_subdag', False)]:
> dag = self._add_dag_attribute(dag, attribute, default)
>
> DAG.sync_to_db(dag, dag.owner, datetime.utcnow())
> dag_pickle = dag.pickle()
>
> dag_model = session.\
> query(DagModel).\
> filter(DagModel.dag_id == dag.dag_id).\
> first()
> dag_model.pickle_id = dag_pickle.id
> dag_model.is_paused = False
>
> session.merge(dag_model)
> session.commit()
>
> bag = DagBag('', include_examples=False)
> bag.bag_dag(dag, None, None)
>
> queue = []
> job = SchedulerJob(dag_id=dag.dag_id)
> job._process_dags(bag, [dag], queue)
>
> for dag_id, task_id, execution_date in queue:
> task = dag.get_task(task_id)
>
> task_instance = TaskInstance(task, execution_date)
> task_instance.state = State.SCHEDULED
>
> session.merge(task_instance)
> session.commit()
>
>
>
> Any help on this workflow would be greatly appreciated!
>
>
> Best,
>
> Georg
>
>
> On Tue, Nov 22, 2016 at 10:46 PM, Maxime Beauchemin <
> maximebeauchemin@gmail.com> wrote:
>
> > I added a paragraph to the FAQ entry: "How can I create DAGs
> dynamically?"
> >
> > The GH link above points straight to that paragraph. It's related to an
> > issue that was discussed on this mailing list last week.
> >
> > I agree that working examples are ideal, and they can help with test
> > coverage as well.
> >
> > Max
> >
> > On Tue, Nov 22, 2016 at 1:10 PM, siddharth anand <sa...@apache.org>
> > wrote:
> >
> > > Hi Max,
> > > Which part in the above PR is related to dynamic dags?
> > >
> > > When thinking about adding documentation about functionality, I propose
> > the
> > > community bias towards adding working examples and test coverage. We
> > offer
> > > a quick start (which by the way needs some updates - for example, why
> > does
> > > it not start airflow-scheduler after starting the webserver?), but then
> > > folks get stuck in how to write DAGs and use the full range of Airflow
> > > capabilities. This is where examples and better test coverage help keep
> > > newbies productive.
> > >
> > > Perhaps the examples and tests can be upgraded to show a fuller set of
> > > dynamic dag capabilities?
> > >
> > > -s
> > >
> > > On Mon, Nov 21, 2016 at 7:55 PM, Maxime Beauchemin <
> > > maximebeauchemin@gmail.com> wrote:
> > >
> > > > I just added a bit of information about dynamic DAG creation here:
> > > > https://github.com/apache/incubator-airflow/pull/1889/files#diff-
> > > > c6f0a0722c6a2f86277535d7bcec7f8cR162
> > > >
> > > > Let me know if it helps.
> > > >
> > > > Max
> > > >
> > > > On Mon, Nov 21, 2016 at 2:58 AM, Deepak Kumar Malladi <
> > > > kapeed2091@gmail.com>
> > > > wrote:
> > > >
> > > > > Hi,
> > > > >
> > > > > I want to dynamically create DAG during run time. I tried the
> snippet
> > > > given
> > > > > in the documentation. But it didnt work for me.
> > > > >
> > > > > Any pointer on how to trigger DAGs which aren't actually present in
> > DAG
> > > > > folder but are created through code execution (dynamically
> created)?
> > > > >
> > > > >
> > > > > Thanks & Regards,
> > > > > Deepak
> > > > >
> > > >
> > >
> >
>