You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2022/07/11 16:21:57 UTC
[GitHub] [airflow] dstandish opened a new pull request, #24969: Move dataset dagrun creation to scheduler main
dstandish opened a new pull request, #24969:
URL: https://github.com/apache/airflow/pull/24969
Depends on https://github.com/apache/airflow/pull/24908/
Moves dataset-triggered dagrun creation to the main scheduler loop so that the _target_ dag can be locked.
Makes the run_id of the dataset-triggered dagrun deterministic.
Adds a table to map dataset events to dagruns.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bbovenzi commented on a diff in pull request #24969: WIP - Move dataset dagrun creation to scheduler main
Posted by GitBox <gi...@apache.org>.
bbovenzi commented on code in PR #24969:
URL: https://github.com/apache/airflow/pull/24969#discussion_r918171500
##########
airflow/migrations/versions/0114_2_4_0_add_dataset_model.py:
##########
@@ -118,17 +118,46 @@ def _create_dataset_dag_run_queue_table():
)
+def _create_dataset_event_table():
+ op.create_table(
+ 'dataset_event',
+ sa.Column('id', Integer, primary_key=True, autoincrement=True),
+ sa.Column('dataset_id', Integer, nullable=False),
+ sa.Column('extra', ExtendedJSON, nullable=True),
+ sa.Column('task_id', String(250), nullable=True),
+ sa.Column('dag_id', String(250), nullable=True),
+ sa.Column('run_id', String(250), nullable=True),
+ sa.Column('map_index', sa.Integer(), nullable=True, server_default='-1'),
+ sa.Column('created_at', TIMESTAMP, nullable=False),
+ sqlite_autoincrement=True, # ensures PK values not reused
+ )
+ op.create_index('idx_dataset_id_created_at', 'dataset_event', ['dataset_id', 'created_at'])
+
+
+def _create_dataset_event_dag_run_table():
+ op.create_table(
+ 'dataset_event_dag_run',
+ sa.Column('dataset_event_id', Integer, nullable=False),
+ sa.Column('dag_run_id', Integer, nullable=False),
Review Comment:
I feel like we also want `dag_id` here, no?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb merged pull request #24969: Move dataset dagrun creation to scheduler main
Posted by GitBox <gi...@apache.org>.
ashb merged PR #24969:
URL: https://github.com/apache/airflow/pull/24969
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] bbovenzi commented on a diff in pull request #24969: WIP - Move dataset dagrun creation to scheduler main
Posted by GitBox <gi...@apache.org>.
bbovenzi commented on code in PR #24969:
URL: https://github.com/apache/airflow/pull/24969#discussion_r918171500
##########
airflow/migrations/versions/0114_2_4_0_add_dataset_model.py:
##########
@@ -118,17 +118,46 @@ def _create_dataset_dag_run_queue_table():
)
+def _create_dataset_event_table():
+ op.create_table(
+ 'dataset_event',
+ sa.Column('id', Integer, primary_key=True, autoincrement=True),
+ sa.Column('dataset_id', Integer, nullable=False),
+ sa.Column('extra', ExtendedJSON, nullable=True),
+ sa.Column('task_id', String(250), nullable=True),
+ sa.Column('dag_id', String(250), nullable=True),
+ sa.Column('run_id', String(250), nullable=True),
+ sa.Column('map_index', sa.Integer(), nullable=True, server_default='-1'),
+ sa.Column('created_at', TIMESTAMP, nullable=False),
+ sqlite_autoincrement=True, # ensures PK values not reused
+ )
+ op.create_index('idx_dataset_id_created_at', 'dataset_event', ['dataset_id', 'created_at'])
+
+
+def _create_dataset_event_dag_run_table():
+ op.create_table(
+ 'dataset_event_dag_run',
+ sa.Column('dataset_event_id', Integer, nullable=False),
+ sa.Column('dag_run_id', Integer, nullable=False),
Review Comment:
I feel like we also want `dag_id` here, no? and then rename the column to `run_id`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dstandish commented on a diff in pull request #24969: WIP - Move dataset dagrun creation to scheduler main
Posted by GitBox <gi...@apache.org>.
dstandish commented on code in PR #24969:
URL: https://github.com/apache/airflow/pull/24969#discussion_r918191487
##########
airflow/migrations/versions/0114_2_4_0_add_dataset_model.py:
##########
@@ -118,17 +118,46 @@ def _create_dataset_dag_run_queue_table():
)
+def _create_dataset_event_table():
+ op.create_table(
+ 'dataset_event',
+ sa.Column('id', Integer, primary_key=True, autoincrement=True),
+ sa.Column('dataset_id', Integer, nullable=False),
+ sa.Column('extra', ExtendedJSON, nullable=True),
+ sa.Column('task_id', String(250), nullable=True),
+ sa.Column('dag_id', String(250), nullable=True),
+ sa.Column('run_id', String(250), nullable=True),
+ sa.Column('map_index', sa.Integer(), nullable=True, server_default='-1'),
+ sa.Column('created_at', TIMESTAMP, nullable=False),
+ sqlite_autoincrement=True, # ensures PK values not reused
+ )
+ op.create_index('idx_dataset_id_created_at', 'dataset_event', ['dataset_id', 'created_at'])
+
+
+def _create_dataset_event_dag_run_table():
+ op.create_table(
+ 'dataset_event_dag_run',
+ sa.Column('dataset_event_id', Integer, nullable=False),
+ sa.Column('dag_run_id', Integer, nullable=False),
Review Comment:
dag_run_id is a key to `dag_run.id` which is a surrogate key for the dag run table (added a couple months back) so we should have everything we need here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a diff in pull request #24969: WIP - Move dataset dagrun creation to scheduler main
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #24969:
URL: https://github.com/apache/airflow/pull/24969#discussion_r934589371
##########
airflow/models/dag.py:
##########
@@ -2991,14 +2991,38 @@ def deactivate_deleted_dags(cls, alive_dag_filelocs: List[str], session=NEW_SESS
continue
@classmethod
- def dags_needing_dagruns(cls, session: Session):
+ def dags_needing_dagruns(cls, session: Session) -> Tuple[Query, Dict]:
"""
Return (and lock) a list of Dag objects that are due to create a new DagRun.
This will return a resultset of rows that is row-level-locked with a "SELECT ... FOR UPDATE" query,
you should ensure that any scheduling decisions are made in a single transaction -- as soon as the
transaction is committed it will be unlocked.
"""
+ from airflow.models.dataset import DatasetDagRef, DatasetDagRunQueue as DDRQ
+
+ # these dag ids are triggered by datasets, and they are ready to go.
+ dataset_triggered_dag_info_list = {
+ x.dag_id: (x.first_event_time, x.last_event_time)
+ for x in session.query(
+ DatasetDagRef.dag_id,
+ func.max(DDRQ.created_at).label('last_event_time'),
+ func.max(DDRQ.created_at).label('first_event_time'),
+ )
+ .join(
+ DDRQ,
+ and_(
+ DDRQ.dataset_id == DatasetDagRef.dataset_id,
+ DDRQ.target_dag_id == DatasetDagRef.dag_id,
+ ),
+ isouter=True,
+ )
Review Comment:
We should define the relationships on the DDRQ table (that way we only have to have it once, not in every place we might want to use this rel) then this can become something like
```suggestion
.join(DatassetDagRef._something_)
```
##########
airflow/jobs/scheduler_job.py:
##########
@@ -997,8 +998,17 @@ def _get_next_dagruns_to_examine(self, state: DagRunState, session: Session):
@retry_db_transaction
def _create_dagruns_for_dags(self, guard, session):
"""Find Dag Models needing DagRuns and Create Dag Runs with retries in case of OperationalError"""
- query = DagModel.dags_needing_dagruns(session)
- self._create_dag_runs(query.all(), session)
+ query, dataset_triggered_dag_info = DagModel.dags_needing_dagruns(session)
+ all_dags_needing_dag_runs = set(query.all())
+ dataset_triggered_dags = [
+ x for x in all_dags_needing_dag_runs if x.dag_id in dataset_triggered_dag_info.keys()
Review Comment:
```suggestion
dag for dag in all_dags_needing_dag_runs if dag.dag_id in dataset_triggered_dag_info
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a diff in pull request #24969: WIP - Move dataset dagrun creation to scheduler main
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #24969:
URL: https://github.com/apache/airflow/pull/24969#discussion_r937939368
##########
airflow/models/dag.py:
##########
@@ -3024,14 +3024,31 @@ def deactivate_deleted_dags(cls, alive_dag_filelocs: List[str], session=NEW_SESS
continue
@classmethod
- def dags_needing_dagruns(cls, session: Session):
+ def dags_needing_dagruns(cls, session: Session) -> Tuple[Query, Dict]:
"""
Return (and lock) a list of Dag objects that are due to create a new DagRun.
This will return a resultset of rows that is row-level-locked with a "SELECT ... FOR UPDATE" query,
you should ensure that any scheduling decisions are made in a single transaction -- as soon as the
transaction is committed it will be unlocked.
"""
+ from airflow.models.dataset import DatasetDagRef, DatasetDagRunQueue as DDRQ
+
+ # these dag ids are triggered by datasets, and they are ready to go.
+ dataset_triggered_dag_info_list = {
+ x.dag_id: (x.first_event_time, x.last_event_time)
+ for x in session.query(
+ DatasetDagRef.dag_id,
+ func.max(DDRQ.created_at).label('last_event_time'),
+ func.max(DDRQ.created_at).label('first_event_time'),
+ )
+ .join(DatasetDagRef.queue_records, isouter=True)
+ .group_by(DatasetDagRef.dag_id)
+ .having(func.count() == func.sum(case((DDRQ.target_dag_id.is_not(None), 1), else_=0)))
+ .all()
+ }
+ dataset_triggered_dag_ids = list(dataset_triggered_dag_info_list.keys())
Review Comment:
I'm slightly worried about how well this will scale in larger installs - i.e. ones with a large number of DAGs (1000+ dags all using datasets)
##########
airflow/models/dag.py:
##########
@@ -3024,14 +3024,31 @@ def deactivate_deleted_dags(cls, alive_dag_filelocs: List[str], session=NEW_SESS
continue
@classmethod
- def dags_needing_dagruns(cls, session: Session):
+ def dags_needing_dagruns(cls, session: Session) -> Tuple[Query, Dict]:
"""
Return (and lock) a list of Dag objects that are due to create a new DagRun.
This will return a resultset of rows that is row-level-locked with a "SELECT ... FOR UPDATE" query,
you should ensure that any scheduling decisions are made in a single transaction -- as soon as the
transaction is committed it will be unlocked.
"""
+ from airflow.models.dataset import DatasetDagRef, DatasetDagRunQueue as DDRQ
+
+ # these dag ids are triggered by datasets, and they are ready to go.
+ dataset_triggered_dag_info_list = {
+ x.dag_id: (x.first_event_time, x.last_event_time)
+ for x in session.query(
+ DatasetDagRef.dag_id,
+ func.max(DDRQ.created_at).label('last_event_time'),
+ func.max(DDRQ.created_at).label('first_event_time'),
+ )
+ .join(DatasetDagRef.queue_records, isouter=True)
+ .group_by(DatasetDagRef.dag_id)
+ .having(func.count() == func.sum(case((DDRQ.target_dag_id.is_not(None), 1), else_=0)))
Review Comment:
What's going on here? This feels like an expensive query to run.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a diff in pull request #24969: Move dataset dagrun creation to scheduler main
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #24969:
URL: https://github.com/apache/airflow/pull/24969#discussion_r946972332
##########
airflow/models/dag.py:
##########
@@ -3024,14 +3024,31 @@ def deactivate_deleted_dags(cls, alive_dag_filelocs: List[str], session=NEW_SESS
continue
@classmethod
- def dags_needing_dagruns(cls, session: Session):
+ def dags_needing_dagruns(cls, session: Session) -> Tuple[Query, Dict]:
"""
Return (and lock) a list of Dag objects that are due to create a new DagRun.
This will return a resultset of rows that is row-level-locked with a "SELECT ... FOR UPDATE" query,
you should ensure that any scheduling decisions are made in a single transaction -- as soon as the
transaction is committed it will be unlocked.
"""
+ from airflow.models.dataset import DatasetDagRef, DatasetDagRunQueue as DDRQ
+
+ # these dag ids are triggered by datasets, and they are ready to go.
+ dataset_triggered_dag_info_list = {
+ x.dag_id: (x.first_event_time, x.last_event_time)
+ for x in session.query(
+ DatasetDagRef.dag_id,
+ func.max(DDRQ.created_at).label('last_event_time'),
+ func.max(DDRQ.created_at).label('first_event_time'),
+ )
+ .join(DatasetDagRef.queue_records, isouter=True)
+ .group_by(DatasetDagRef.dag_id)
+ .having(func.count() == func.sum(case((DDRQ.target_dag_id.is_not(None), 1), else_=0)))
+ .all()
+ }
+ dataset_triggered_dag_ids = list(dataset_triggered_dag_info_list.keys())
Review Comment:
This only affects people using Datasets, which is a new feature, so this can be fixed later if needed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org