Create a more efficient airflow dag test command that also has better local logging#26400
Conversation
|
This looks sooooper cool :). I will do some quick testing to see the experience. One suggestions (asking more what you think about it and maybe those could be separate PRs., or maybe we could incorporate it in this one). How about adding integration with I imagine just adding WDYT @dimberman ? |
…roved-local-dag-test
e295ce7 to
f2df9cf
Compare
| log.info("*****************************************************") | ||
| log.info("Running task %s", current_task.task_id) | ||
| try: | ||
| xcom_value = current_task.execute(context=ti.get_template_context()) |
There was a problem hiding this comment.
| xcom_value = current_task.execute(context=ti.get_template_context()) | |
| xcom_value = ti.task.execute(context=ti.get_template_context(session=session)) |
There was a problem hiding this comment.
You also aren't calling pre_execute or post_execute here, so some operators might not work.
You probaly want to call a method on TI instead.
Possible ti._run_raw_task(session=session, test_mode=True)
There was a problem hiding this comment.
@ashb I agree on using the _run_raw_task function, but ironically the test_mode might not be the best idea here. test_mode prevents us from recording passing/failure in the database, and without that state we won't know when to run downstream tasks.
There was a problem hiding this comment.
That's not quite what test mode does.
It simply doesn't a session.commit(). But if you put a session.flush() after it (and are careful about transactions/passing session everywhere) then you can run all tasks in a single transaction, never commit it, and roll it back at the end.
I just tested this change and it works for downstream. See the video below.
There was a problem hiding this comment.
@ashb when I run this with test mode it seems to infinite loop as the tasks are never marked as finished?
|
Thank you @potiuk ! Do you think that would be too oppinionated to choose a specific debugger? One really awesome thing we could potentially do with this is even write a guide to use this in pycharm and vscode (since there's no parallelism it shouldn't be hard) |
|
I don't think we should integrate any specific debugger here in this CLI function, but instruct people to use |
I think not - especially that it is console textui/console only . I honestly don't care what the debugger is, I want this to run effortlessly. If we can make this works out-of-the-box without any manual setup, I would be all for it. The problem with vscode/Pycharm is that it is not at all straightforward to integrate with "airflow dag test" command. Have you tried to add debug mode for any "airflow" command in either of them, this is a lot of hassle. First of all - it will not work when you have remote /shell access easily. Secondly, you cannot easily turn "airflow" command iine into a debuggable run. You need to create a custom configuraiton first, then have a local venv configured with the right version of airlfow installed and configured in the IDE, third you need to do "run -> create debug configuration" then you have to choose "module" then choose airflow module, then type extra parameters to pass. That's a lot of keystrokes and knowledge what to do. AND you have to heave your IDE installed in the place where you have your Airlfow installed. This might be acceptable for Breeze/Airflow development, but not at all acceptable for DAG development. Using "opinionated debugger" with console interface is so much less friction. just 'airflow dag test test" and add Let me reverse the question: What is the problem you foresee by having an opinionated predefined flag when it does not close any of the other options? |
|
We can even do not make it a requiremeent, but install it dynamically when someone uses |
|
Or name it |
|
It needs a bit of work/polish, but using this That gives me something I can run against a dag file. To make that work I did add this to the dag file (and a couple of other local changes to make this work for if __name__ == "__main__":
dag.cli()Edit: THis works from VScode running on windows against an airflow in WSL2. |
|
Sneak peak 2022-09-14.20-34-13.webm |
We already have DebugExecutor that does exactly that in the same way and a bit simpler (and is portable across IDEs) I think.. https://airflow.apache.org/docs/apache-airflow/stable/executor/debug.html?highlight=debug+executor I was telling about debugging without having any IDE setup. |
| dr: DagRun = ( | ||
| session.query(DagRun) | ||
| .filter(DagRun.dag_id == dag.dag_id, DagRun.execution_date == execution_date) | ||
| .first() | ||
| ) | ||
| if dr: | ||
| session.delete(dr) | ||
| session.commit() |
There was a problem hiding this comment.
It might make sense to create _delete_dagrun for that. Mainly because of the low-level session calls imo. WDYT?
There was a problem hiding this comment.
Do we foresee other use-cases to delete dagruns? FWIW I believe @ashb plans to look into a way for us to do this without ever actually touching the DB which would make this irrelevant in the long run.
Co-authored-by: Felix Uellendall <feluelle@users.noreply.github.com>
…roved-local-dag-test
Co-authored-by: Felix Uellendall <feluelle@users.noreply.github.com>
Co-authored-by: Felix Uellendall <feluelle@users.noreply.github.com>
…roved-local-dag-test
…roved-local-dag-test
…roved-local-dag-test
|
|
||
| Debug Executor | ||
| ================== | ||
| Debug Executor (deprecated) |
There was a problem hiding this comment.
Does this mean we want to remove DebugExecutor in Airflow 3?
There was a problem hiding this comment.
+1 to @eladkal question.
I stumbled across this while working on something entirely different. I'm very curious to have this discussion as well.
Currently AIP-47 system tests depend on the DebugExecutor:
The tests should be structured in the way that they are easy to run as “standalone” tests manually but they should also nicely be integrated into pytest test execution environment. This can be achieved by leveraging the DebugExecutor and utilising modern pytest test discovery mechanism
So we'd need to do some migration there if we were to deprecate this executor (CC @potiuk @mnojek). I know @bhirsz has mentioned before they're using a different executor altogether for their system tests, maybe that's a more viable option.
We should also add deprecation warnings if we plan to deprecate this executor.
Cleanup the docstrings which were still mentioning the usage of the DebugExecutor
The current
airflow dag testcommand is insufficient for an iterative local workflow when developing dags. The current system relies on running a backfill job which a) is extremely slow and b) only shows scheduler-level logs, so the user would need to search for log files to see how their tasks actually ran!This new system
Speed
With this new system, we see a notable speedup in between-task speed. The new airflow dag test command can load tasks immediately and not wait for a scheduler heartbeat or other irrelevant airflow processes.
Before:
example bash operator: 13 seconds
example python operator: 26 seconds
After
example bash operator: 6 seconds
example python operator: 16 seconds
Logging
Previously, the
airflow dag testcommand would pass on scheduler-level logs. This gives essentially 0 useful information for a DAG writer, and would force them to seek out task.log files in their filesystem every time they run a test dag. To improve this, we now surface task-level logs to the command line. We also no longer have any scheduler-level logs as we are no longer using a scheduler!Before
After:
^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named
{pr_number}.significant.rstor{issue_number}.significant.rst, in newsfragments.