-
Notifications
You must be signed in to change notification settings - Fork 17.4k
Remove DAG parsing from StandardTaskRunner #26750
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -47,7 +47,6 @@ | |
| from airflow.utils import cli as cli_utils | ||
| from airflow.utils.cli import ( | ||
| get_dag, | ||
| get_dag_by_deserialization, | ||
| get_dag_by_file_location, | ||
| get_dag_by_pickle, | ||
| get_dags, | ||
|
|
@@ -364,14 +363,7 @@ def task_run(args, dag=None): | |
| print(f'Loading pickle id: {args.pickle}') | ||
| dag = get_dag_by_pickle(args.pickle) | ||
| elif not dag: | ||
| if args.local: | ||
| try: | ||
| dag = get_dag_by_deserialization(args.dag_id) | ||
| except AirflowException: | ||
| print(f'DAG {args.dag_id} does not exist in the database, trying to parse the dag_file') | ||
| dag = get_dag(args.subdir, args.dag_id) | ||
| else: | ||
| dag = get_dag(args.subdir, args.dag_id) | ||
| dag = get_dag(args.subdir, args.dag_id, include_examples=False) | ||
|
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Sorry for this late reply. Can you share the test case?
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. diff --git a/airflow/cli/commands/task_command.py b/airflow/cli/commands/task_command.py
index 982aa31fd5..ab05bb56a9 100644
--- a/airflow/cli/commands/task_command.py
+++ b/airflow/cli/commands/task_command.py
@@ -578,13 +578,13 @@ def task_render(args):
task, args.map_index, exec_date_or_run_id=args.execution_date_or_run_id, create_if_necessary="memory"
)
ti.render_templates()
- for attr in task.__class__.template_fields:
+ for attr in task.template_fields:
print(
textwrap.dedent(
f""" # ----------------------------------------------------------
# property: {attr}
# ----------------------------------------------------------
- {getattr(task, attr)}
+ {getattr(ti.task, attr)}
"""
)
)
diff --git a/tests/cli/commands/test_task_command.py b/tests/cli/commands/test_task_command.py
index c565f601d7..f3673fb2dd 100644
--- a/tests/cli/commands/test_task_command.py
+++ b/tests/cli/commands/test_task_command.py
@@ -388,6 +388,54 @@ class TestCliTasks:
assert 'echo "2016-01-01"' in output
assert 'echo "2016-01-08"' in output
+ def test_mapped_task_render(self):
+ """
+ tasks render should render and displays templated fields for a given mapped task
+ """
+ with redirect_stdout(io.StringIO()) as stdout:
+ task_command.task_render(
+ self.parser.parse_args(
+ [
+ "tasks",
+ "render",
+ "test_mapped_classic",
+ "consumer_literal",
+ "2022-01-01",
+ "--map-index",
+ "0",
+ ]
+ )
+ )
+ # the dag test_mapped_classic has op_args=[[1], [2], [3]], so the first mapped task should have op_args=[1]
+ output = stdout.getvalue()
+ assert "[1]" in output
+ assert "[2]" not in output
+ assert "[3]" not in output
+ assert "property: op_args" in output
+
+ def test_mapped_task_render_with_template(self):
+ """
+ tasks render should render and displays templated fields for a given mapped task
+ """
+ with redirect_stdout(io.StringIO()) as stdout:
+ task_command.task_render(
+ self.parser.parse_args(
+ [
+ "tasks",
+ "render",
+ "test_mapped_task_with_template",
+ "some_command",
+ "2022-01-01",
+ "--map-index",
+ "0",
+ ]
+ )
+ )
+ # the dag test_mapped_classic has op_args=[[1], [2], [3]], so the first mapped task should have op_args=[1]
+ output = stdout.getvalue()
+ assert 'echo "2022-01-01"' in output
+ assert 'echo "2022-01-08"' in output
+
def test_cli_run_when_pickle_and_dag_cli_method_selected(self):
"""
tasks run should return an AirflowException when invalid pickle_id is passed
diff --git a/tests/dags/test_mapped_task_with_templates.py b/tests/dags/test_mapped_task_with_templates.py
new file mode 100644
index 0000000000..fe0b8d0c49
--- /dev/null
+++ b/tests/dags/test_mapped_task_with_templates.py
@@ -0,0 +1,36 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import datetime
+from textwrap import dedent
+
+from airflow import DAG
+from airflow.operators.bash import BashOperator
+
+with DAG(dag_id="test_mapped_task_with_template", start_date=datetime.datetime(2022, 1, 1)) as dag:
+ templated_command = dedent(
+ """
+ {% for i in range(5) %}
+ echo "{{ ds }}"
+ echo "{{ macros.ds_add(ds, 7)}}"
+ {% endfor %}
+ """
+ )
+ commands = [templated_command, "echo 1"]
+
+ BashOperator.partial(task_id="some_command").expand(bash_command=commands)
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Thanks @ephraimbuddy for the patch. I will use it and also add you as co-author. |
||
| else: | ||
| # Use DAG from parameter | ||
| pass | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
hi @ephraimbuddy could you please give more context about why this should be removed?
it looks like your PR only tried to optimize when executing a task via fork. there is also an important part https://github.com/apache/airflow/pull/26750/files#diff-8b336b0706abbada03e8dce519e122384da2c913cc79ee4d638d0427a1342d41L47-L48
i think we should need
dag = get_dag_by_deserialization(args.dag_id)please let me know your thoughtsThere was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We noticed a significant increase in task run duration with the previous change. Using
get_dag_by_deserializationonly delays the dag parsing(for_start_by_fork). Looking atexecI think we should devise another way to get the serialized dag for use because using the previous change, we had to parse the dag in TaskRunner process forforkwhich causes delays.