SparkKubernetesOperator is not working when we are passing YAML string

[xs2tarunkukreja]

Apache Airflow version

2.7.2

What happened

Currently I am using -
Python version: 3.8.18
Airflow version: 2.7.1v (on K8S)

When we run SparkKubernetesOperator function with YAML file path, it work fine. But when we are running the same after reading the YAML file, it fails.

Example -

def load_template_v1():
    ....

    with open(template_path) as pyspark_template_file:
        template = yaml.safe_load(pyspark_template_file)

    return template

What you think should happen instead

It should launch Spark Operator successfully.

How to reproduce

Instead of passing YAML file, pass the content of file.

Operating System

K8S

Versions of Apache Airflow Providers

Python version: 3.8.18
Airflow version: 2.7.1

Deployment

Official Apache Airflow Helm Chart

Deployment details

HELM

Python version: 3.8.18
Airflow version: 2.7.1

Anything else

No

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[Taragolis]

template = yaml.safe_load(pyspark_template_file)

This operation transform into the python objects, e.g. it could be dict, lists, and others.
The difficult to say in which attribute you tried to provide this output and is it even documented / expected.

Could you provide a bit more details, e.g. example of your tasks/dags

[xs2tarunkukreja]

I am just creating a spark job. When I pass file name as application_file = "pipeline.yaml" in SparkKubernetesOperator, it works fine. But when I pass the yaml as string e.g. application_file = load_template_v1(), it start to fail.

I am getting following error -
[2023-10-17 10:23:24,375] Task failed with exception
Traceback (most recent call last):
File "/home/airflow/.local/lib/python3.8/site-packages/airflow/providers/cncf/kubernetes/operators/spark_kubernetes.py", line 114, in execute
body = _load_body_to_dict(self.application_file)
File "/home/airflow/.local/lib/python3.8/site-packages/airflow/providers/cncf/kubernetes/hooks/kubernetes.py", line 46, in _load_body_to_dict
body_dict = yaml.safe_load(body)
File "/home/airflow/.local/lib/python3.8/site-packages/airflow/utils/yaml.py", line 46, in safe_load
return orig(stream, SafeLoader)
File "/home/airflow/.local/lib/python3.8/site-packages/yaml/init.py", line 79, in load
loader = Loader(stream)
File "/home/airflow/.local/lib/python3.8/site-packages/yaml/cyaml.py", line 26, in init
CParser.init(self, stream)
File "yaml/_yaml.pyx", line 288, in yaml._yaml.CParser.init
TypeError: a string or stream input is required

[xs2tarunkukreja]

It is working fine with older version of airflow.

[Taragolis]

That is not a part of Airflow core, this part of Community provider apache-airflow-providers-cncf-kubernetes.

According provided stacktrace, you have a problem with application_file attribute, however docstring said that it expected path to the YAML/JSON (as part of Airflow templates) or YAML/JSON string.

That mean previously it accidentally work as undocumented feature, however stop working after some release, I assume after this one #29977.

Feel free to open a feature request, or if you know how to add this functionality feel free to open a PR directly:

• Contributor's Quick Start
• Airflow Breeze
• TESTING
• Static code checks

[xs2tarunkukreja]

however [docstring] said that it expected path to the YAML/JSON (as part of Airflow templates) or YAML/JSON string.

So, we can pass YAML string. Initially we are passing YAML file path and in second option, we are passing file content means YAML string.

So, it was not accidental thing. Would you please share the community link, where I should ask this question?

[Taragolis]

So, it was not accidental thing. Would you please share the community link, where I should ask this question?

Community providers also developed in this repo. So you could ask questions in Github Discussions or in Slack (more suitable channels are random or troubleshooting)

My main point, and why I close this issue and transform into the discussion, this part previously works just for some reason, but it not documented, so it can't be classified as a Bug.

In the other hand it might be useful to add this feature into the Kubernetes Provider in this case this behaviour should be documented and tested.

[Taragolis]

Other thing which might help you, it is provide to the Operator what it expected: string representation of YAML

In this case you need to change your code to

def load_template_v1():
    ....

    with open(template_path) as pyspark_template_file:
        return pyspark_template_file.read()