Delete old Spark Application in SparkKubernetesOperator

[theesen]

SparkKubernetesOperator deletes any previous Spark Application with the same name

Current issue:
If a Spark App is being launched you either have to template the name inside the yaml file (making it unique with a timestamp) or need to delete the previous run of the Spark App in order to prevent failure. Especially for newcomers to the topic this quickly leads to errors by design of the operator.

Open Questions:
I am wondering if we should delete the K8 Spark Application after a successful run as well from inside the Operator. Or at least provide a flag to enable this.
Downsides: Harder to debug since container is gone in case of errors (or we keep the container if it failed)
Upsides: Less polluted K8 Cluster with old Spark Apps.

I would love to get some input on these thoughts.

Minor Updates:
As far as I could tell the docstring was not correct. You can only pass a String or Dict as application_file . The name of the parameter would need to be changed from my point of view as well. But since this is my first PR here, I did not want to introduce a breaking change to an Operator right away. But if someone makes sure I am doing this the right way, I am happy to adjust.

closes: #16290

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst)
Here are some useful points:

• Pay attention to the quality of your code (flake8, mypy and type annotations). Our pre-commits will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it’s a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[jherrmannNetfonds]

Hi @theesen, thank you for that PR.
I implemented the option with the flag to delete after successful run which you would like to discuss. But I implemented this in SparkKubernetesSensor
I delete the application on success state by default. Since if everything runs fine I do not need keep the application around. Or at least I never had the use case to look into the application.
This way I only keep failed applications and on a rerun of a successful run I do not get any error.

Independent of the implementation I would offer a flag so that the user can choose what behavior he wants.

[potiuk]

Could you please add tests to the change @theesen ?

[potiuk]

Tests failing

[theesen]

Fixed and tested the unittests locally. Should be fine now. Finger crossed.

[potiuk]

Need rebase unfortunately.

[theesen]

@potiuk can you approve the workflow?

[boring-cyborg]

Awesome work, congrats on your first merged pull request!

[assaf-xm]

@theesen , @potiuk , were the changes in this PR reverted in 9a4f674 ?

I can see that with the latest airflow version (2.6.2), existing sparkapplication with the same name won't be deleted by the SparkKubernetesOperator logic, and will fail to launch the new application until manually deleting the old one. Is there any flag to bring back the auto deletion logic?

[potiuk]

No idea. But you can check the code, or with the author of the change you mention - the author might be the best to discuss the change with.

[theesen]

Hey,
yeah the author of the commit linked, refactored the sequence a bit and moved the call of delete_custom_object into the on_kill method. Which is ok in failure cases but does not solve the original intention of my implementation. Which was removing previous runs with the exact same name. Fixing this again should be easy though.

self.hook.delete_custom_object needs to be called in the execute block again. Initially the approach was just to try to delete a spark app with that name, and catch the exception, if it does not exist as an expected case.