You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "patgarz (via GitHub)" <gi...@apache.org> on 2023/09/08 01:09:22 UTC
[GitHub] [airflow] patgarz opened a new pull request, #34198: Add instructions for dynamically extending operators
patgarz opened a new pull request, #34198:
URL: https://github.com/apache/airflow/pull/34198
<!--
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.
-->
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of an existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
#33735 added docs of how to extend, specifically, the `template_ext` attribute of operators. This PR suggests a change to extend that language to encompass the capability to dynamically set operator attributes more generally, and also adds an example DAG for the same.
Concerns:
- While useful, this feels like a footgun in a lot of ways. Should there be some warning text along the lines of "be sure you know what you're doing"?
- The third case in the example DAG, where a DAG author could impact operator execution across other DAGs, was surprising and somewhat worrisome
<!-- Please keep an empty line above the dashes. -->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [newsfragments](https://github.com/apache/airflow/tree/main/newsfragments).
--
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] potiuk commented on pull request #34198: Add instructions for dynamically extending operators
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #34198:
URL: https://github.com/apache/airflow/pull/34198#issuecomment-1713635698
> If you _really_ want to just change the operator in this one DAG, the way to do it is
>
> ```python
> op = Operator(...)
> op.template_ext = (*op.template_ext, "another_field")
> ```
>
> Patching the operator class _in a DAG_ is just very much no. It actually makes some sense if you share that modified class across multiple DAG files, but expecting a patched class to only affect the current module will lead you into a hell constructed of side effects, as illustrated above.
Yep. That's more isolated and likely we should suggest to do it instead.
--
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] potiuk commented on pull request #34198: Add instructions for dynamically extending operators
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #34198:
URL: https://github.com/apache/airflow/pull/34198#issuecomment-1717958532
> The patch does not leak because of Airflow’s internal implementation, a worker only loads one DAG file that the task is defined in. If this is done in an imported module, the patch leaks into every file that imports the module (obviously), and if a file defines multiple DAGs, it leaks into all the DAGs in the same file (less obviously). The difference being subtle is why I’m not exactly sure it’s a good idea to recommend the approach in examples.
Agree. The confusion on where the leaks might happen is too dangerous. That's why "per task" overried is better than "per class".
--
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] patgarz commented on pull request #34198: Add instructions for dynamically extending operators
Posted by "patgarz (via GitHub)" <gi...@apache.org>.
patgarz commented on PR #34198:
URL: https://github.com/apache/airflow/pull/34198#issuecomment-1716389975
I agree with all points made. Modifying instances of the operator (seems to) work fine, whereas patching the operator itself has unexpected and difficult to predict side-effects. That's the same recommendation I settled on in the example DAG:
https://github.com/apache/airflow/pull/34198/files#diff-3500539446f798375443351a94a3942c6df0211914382a0ecdf6954829bf3355R59
And while writing this I realize the language I used in the suggested documentation itself wasn't clear. I've updated that language to very much recommend against patching the actual operator.
--
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] potiuk commented on pull request #34198: Add instructions for dynamically extending operators
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #34198:
URL: https://github.com/apache/airflow/pull/34198#issuecomment-1716401383
cc: @uranusjr ?
--
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
Re: [PR] Add instructions for dynamically extending operators [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #34198:
URL: https://github.com/apache/airflow/pull/34198#issuecomment-2144074048
This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.
--
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] Taragolis commented on pull request #34198: Add instructions for dynamically extending operators
Posted by "Taragolis (via GitHub)" <gi...@apache.org>.
Taragolis commented on PR #34198:
URL: https://github.com/apache/airflow/pull/34198#issuecomment-1711521794
> Actually I think this was a discovery we made recently, that this is possible and our users CAN do it this way.
Personally I use this approach since Airflow 1.10.4 🤣 . I don't want to say that this correct way and for sure that approach has side effects, which related how class attributes works in Python. So DAG Authors should actually know what and why they do that
```python
# module1
class A:
a = ("a", "b")
```
```python
# module2
from module1 import A
A.a = ("c", "d")
class B:
...
```
```python
# module3
from module1 import A
a = A()
assert a.a == ("a", "b"), f"Before import module2 {a.a}"
from module2 import B
assert a.a == ("a", "b"), f"After import module2 {a.a}"
# AssertionError: After import module2 ('c', 'd')
```
--
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] potiuk commented on pull request #34198: Add instructions for dynamically extending operators
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #34198:
URL: https://github.com/apache/airflow/pull/34198#issuecomment-1716641785
(As a sidenote, as expected, using core execute_tasks_new_python_interpreter does seem to make it so DAG1 doesn't interfere with DAG2, unless they directly import each other. Not sure if that's worth adding here or somewhere else or not at all given it's a pretty extreme edge case.)
```
if settings.EXECUTE_TASKS_NEW_PYTHON_INTERPRETER:
state = self._execute_work_in_subprocess(command)
else:
state = self._execute_work_in_fork(command)
```
I think even `False` does it. The way it works if it is set to Fa;se - we are FORKING the main interpreter before we run task. This means that any change you made in the class will not leak to any other task (because parsing the the DAG happens after the fork happens). So there is no risk whatsoever it will leak. That was my original point.
However, I agree, it's counter-intuitive. You have to know that we do it, you have to know what forking is and when it is done, and why any changes to class will not leak to another forrk or to the parent process. All this might be surprising to anyone who is used to regular Python running in regular interpreter, because they expect class change will affect everything else. That's why I agree (and this is the only reason - potential user's confusion here) why we shoud only modify the operators. There is no risk of leaking stuff if we modify classes, but there is a potential user confusion.
--
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] potiuk commented on pull request #34198: Add instructions for dynamically extending operators
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #34198:
URL: https://github.com/apache/airflow/pull/34198#issuecomment-1711245962
> I’m not sure I’d want my colleagues to do this. The standard way to extending a class is to subclass, not monkey patching something on the class.
Actually I think this was a discovery we made recently, that this is possible and our users CAN do it this way. And I believe `many` users of ours find this as much easier way of changing definition of templated fields rather than extending classes (and it can be extended to other class fields without side-effect or dangers - I think).
I personally find it as a very smart, simple and perfectly fine way of changing template fields. But maybe there are some drawbacks/side-effects I could not see.
I believe it should have no side effect and should be a very straightforward way of changing the set of templated_fields and template_ext especially (that was always a pain when someone **just** missed one field in those or where some operator had a `template_ext` configured to be `.json` but one of your fields had to have field that was ending with .json (see https://github.com/apache/airflow/issues/33694).
The "Literally" idea in #33694 @uranusjr idea is cool to solve "template_ext" problem, and we might implement it to solve it problem in 2.8.0. But this one has a solution for pre-2.8.0 for that.
However, many users have been struggling with fields in template_ext, and we even toyed with the possibility of making all fields templated (which is not possible for back-compatibility reasons) - and we've always suggested to create custom classes for that.
But the need to define a new class for `template_ext` is rather cumbersome for this case and has some ripple effects.
Compare:
``
Operator.template_ext = (*Operator.template_ext, "another_field")
``
vs.
```
class MyOperator(Operator):
template.ext = (*Operator.template_ext, "another_field")
```
+ (and this is the difficult part) you need to change to use MyOperator in all your DAGs.
Defining your own 'MyOperatora` class has also the drawback, that if you have some common utilities imported that create DAGs or parts of your DAGs, you will actually need to move the "MyOperator" to those common utiltities to use it and then if you would like to change the `template_ext` only in one or two of your DAGs, then it becomes very very messy.
The monkeypatching solution is IMHO much cleaner, more elegant - and avoids the "common utils" mess - because we **guarantee** all DAGs are parsed in isolated subprocess, the scope of monkeypatching of the Operator is limited to single dag file. Full stop. It does not leak anywhere, there are no side-effects. Once the DAG is serialized, the operator's fields are marked as templated according to monkeypatched list and the webserver and all other components will just use the serialized version. I cannot find how this could cause other side effect.
I think it's quite cool to suggest it to the users that they can do it (for sure for `template_ext` and `template_fields`). And it can be extended to other class fields as well (as suggested here) - the monkeypatching effect is really limited to "one dag file" and this is guaranteed by the way how DAGs are parsed and how tasks are executed and how all the other components use just serialized version of the DAG.
For now I don't see how this might be problematic, and it seems like an elegant way to solve quite a number of pain points our users have in a very elegant way. But I would love to hear what others think. Adding the usual suspects for these discussions (maybe I missed someone):
@jedcunningham @dstandish @Taragolis @kaxil @pankajastro @o-nikolas @hussein-awala @eladkal @ephraimbuddy
--
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] patgarz commented on pull request #34198: Add instructions for dynamically extending operators
Posted by "patgarz (via GitHub)" <gi...@apache.org>.
patgarz commented on PR #34198:
URL: https://github.com/apache/airflow/pull/34198#issuecomment-1719757689
Sure, I think my point is that even without the case you talked about (cross-import statements among DAGs), I was seeing leakage from a file defined in `example_dags` even though the exact same file didn't have leakage when added to `files/dags`.
I'm not *too* worried about this as I wasn't able to replicate the leakage (as expected, as you note) with any other case other than `example_dags` but if we don't know why the `example_dags` is leaky then I think that's something to figure out if it's replicable to a different case than `example_dags`.
Or maybe you've tested yourself, didn't see the same, and I've made repeated errors as I test this case, in which case, I will shut up and move on 😄
--
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] uranusjr commented on pull request #34198: Add instructions for dynamically extending operators
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on PR #34198:
URL: https://github.com/apache/airflow/pull/34198#issuecomment-1711196083
I’m not sure I’d want my colleagues to do this. The standard way to extending a class is to subclass, not monkey patching something on the class.
--
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] patgarz commented on pull request #34198: Add instructions for dynamically extending operators
Posted by "patgarz (via GitHub)" <gi...@apache.org>.
patgarz commented on PR #34198:
URL: https://github.com/apache/airflow/pull/34198#issuecomment-1716416870
(As a sidenote, using `core execute_tasks_new_python_interpreter` *does* seem to make it so DAG1 doesn't interfere with DAG2, unless they directly import each other.)
--
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] uranusjr commented on pull request #34198: Add instructions for dynamically extending operators
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on PR #34198:
URL: https://github.com/apache/airflow/pull/34198#issuecomment-1711549102
If you _really_ want to just change the operator in this one DAG, the way to do it is
```python
op = Operator(...)
op.template_ext = (*op.template_ext, "another_field")
```
Patching the operator instance is just very much no. It actually makes some sense if you share that modified class across multiple DAG files, but expecting a patched class to only affect the current module will lead you into a hell constructed of side effects.
--
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] potiuk commented on pull request #34198: Add instructions for dynamically extending operators
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #34198:
URL: https://github.com/apache/airflow/pull/34198#issuecomment-1716400752
Looks cool :)
--
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] potiuk commented on pull request #34198: Add instructions for dynamically extending operators
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #34198:
URL: https://github.com/apache/airflow/pull/34198#issuecomment-1719763160
The example_dags are different. They are imported directly from airflow package, not from DAG folder, so they are not parsed by DAGFileProcessor - they are parsed inside the Airflow Scheduler
--
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] patgarz commented on pull request #34198: Add instructions for dynamically extending operators
Posted by "patgarz (via GitHub)" <gi...@apache.org>.
patgarz commented on PR #34198:
URL: https://github.com/apache/airflow/pull/34198#issuecomment-1716947640
Very interesting -- via Breeze, I am actually seeing leakage between DAGs, but *only* when the operator class is modified in `airflow/example_dags` (for example, if I switch the comments in the last section to use the "don't use this method". If I load two files in `files/dags` without the "wrong" example DAG method, I see no leakage.
Are DAGs in `example_dags` loaded differently in some way?
Test DAG:
```py
import pendulum
from airflow.decorators import dag
from airflow.operators.smooth import SmoothOperator
@dag(
dag_id="test1",
schedule_interval=None,
start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
catchup=False,
tags=["test"],
)
def dynamic_operator_taskflow():
not_so_smooth_operator = SmoothOperator(task_id="not_so_smooth")
dynamic_operator_taskflow()
```
Last task in example_dynamic_operator_attributes changed to:
```py
SmoothOperator.yt_link = "https://www.youtube.com/watch?v=dQw4w9WgXcQ"
smooth_operator = SmoothOperator(task_id="not_so_smooth")
```
Result log for `test1/not_so_smooth`:
```txt
b01e16aba829
*** Found local files:
*** * /root/airflow/logs/dag_id=test1/run_id=manual__2023-09-13T04:58:34.976323+00:00/task_id=not_so_smooth/attempt=1.log
[2023-09-13, 04:58:36 UTC] {taskinstance.py:1164} INFO - Dependencies all met for dep_context=non-requeueable deps ti=<TaskInstance: test1.not_so_smooth manual__2023-09-13T04:58:34.976323+00:00 [queued]>
[2023-09-13, 04:58:36 UTC] {taskinstance.py:1164} INFO - Dependencies all met for dep_context=requeueable deps ti=<TaskInstance: test1.not_so_smooth manual__2023-09-13T04:58:34.976323+00:00 [queued]>
[2023-09-13, 04:58:36 UTC] {taskinstance.py:1366} INFO - Starting attempt 1 of 1
[2023-09-13, 04:58:36 UTC] {taskinstance.py:1387} INFO - Executing <Task(SmoothOperator): not_so_smooth> on 2023-09-13 04:58:34.976323+00:00
[2023-09-13, 04:58:36 UTC] {standard_task_runner.py:60} INFO - Started process 1110 to run task
[2023-09-13, 04:58:36 UTC] {standard_task_runner.py:87} INFO - Running: ['***', 'tasks', 'run', 'test1', 'not_so_smooth', 'manual__2023-09-13T04:58:34.976323+00:00', '--job-id', '10', '--raw', '--subdir', 'DAGS_FOLDER/test.py', '--cfg-path', '/tmp/tmpaimkgfy5']
[2023-09-13, 04:58:36 UTC] {standard_task_runner.py:88} INFO - Job 10: Subtask not_so_smooth
[2023-09-13, 04:58:36 UTC] {task_command.py:420} INFO - Running <TaskInstance: test1.not_so_smooth manual__2023-09-13T04:58:34.976323+00:00 [running]> on host b01e16aba829
[2023-09-13, 04:58:37 UTC] {taskinstance.py:1667} INFO - Exporting env vars: AIRFLOW_CTX_DAG_OWNER='***' AIRFLOW_CTX_DAG_ID='test1' AIRFLOW_CTX_TASK_ID='not_so_smooth' AIRFLOW_CTX_EXECUTION_DATE='2023-09-13T04:58:34.976323+00:00' AIRFLOW_CTX_TRY_NUMBER='1' AIRFLOW_CTX_DAG_RUN_ID='manual__2023-09-13T04:58:34.976323+00:00'
[2023-09-13, 04:58:37 UTC] {smooth.py:38} INFO - Enjoy Sade - Smooth Operator: https://www.youtube.com/watch?v=dQw4w9WgXcQ
[2023-09-13, 04:58:37 UTC] {taskinstance.py:1405} INFO - Marking task as SUCCESS. dag_id=test1, task_id=not_so_smooth, execution_date=20230913T045834, start_date=20230913T045836, end_date=20230913T045837
[2023-09-13, 04:58:37 UTC] {local_task_job_runner.py:234} INFO - Task exited with return code 0
[2023-09-13, 04:58:37 UTC] {taskinstance.py:2781} INFO - 0 downstream tasks scheduled from follow-on schedule check
```
Again, if I do this differently (instead with a test1 and test2 dag in files/dags, where I set `SmoothOperator.yt_link = "https://www.youtube.com/watch?v=dQw4w9WgXcQ"`, there does not seem to be leakage (as I would expect based on your comment).
--
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] uranusjr commented on pull request #34198: Add instructions for dynamically extending operators
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on PR #34198:
URL: https://github.com/apache/airflow/pull/34198#issuecomment-1717026010
The patch does not leak because of Airflow’s internal implementation, a worker only loads one DAG file that the task is defined in. If this is done in an imported module, the patch leaks into every file that imports the module (obviously), and if a file defines multiple DAGs, it leaks into all the DAGs in the same file (less obviously). The difference being subtle is why I’m not exactly sure it’s a good idea to recommend the approach in examples.
--
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] potiuk commented on pull request #34198: Add instructions for dynamically extending operators
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #34198:
URL: https://github.com/apache/airflow/pull/34198#issuecomment-1715838404
@patgarz - WDYT?
--
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