Integrating the Kubeflow Spark Application with the Kubeflow SDK

[CrashCumber]

What you would like to be added?

Hello,
I was wondering if there are any plans for integrating with Kubeflow Spark Applications? It would be very helpful to have a unified solution for Python, so that we can run Spark tasks from the pipeline with minimal interaction with the Kubernetes API, customize the start of jobs, and monitor their progress.

Why is this needed?

Spark is an important tool for working with data in the machine learning pipeline. The Kubeflow ecosystem includes a Spark solution, but there is currently no tool available to easily integrate and use it within the Kubeflow pipeline.

Love this feature?

Give it a 👍 We prioritize the features with most 👍

[andreyvelich]

Thanks for creating this @CrashCumber!
Yes, this is in our ROADMAP. We are discussing with @Shekharrajak to add support for SparkApplication CRD with SparkClient() using Kubeflow SDK.
Here is the draft KEP: https://docs.google.com/document/d/1l57bBlpxrW4gLgAGnoq9Bg7Shre7Cglv4OLCox7ER_s/edit?tab=t.0
Recordings from our last call: https://youtu.be/vkijVRB98HA?t=917

We've been talking about two approaches:

Option 1: Using Kubeflow SDK to submit SparkApplication CRD, and manage it via public APIs (e.g. get_job(), list_job(), get_job_logs(), etc.) Similar to how we manage TrainJob with TrainerClient()

Option 2: Using Kubeflow SDK to connect to the existing Spark cluster which is deployed on Kubernetes, and submit Spark jobs there. I guess, similar to how AWS EMR manages Spark clusters.

I am more in favour of Option 1, since Spark already has Spark connection option for that. Additionally, in the future we can expand our SparkClient() to allow connection to the existing Spark clusters.

@CrashCumber Please can you share more about your use-cases and how you think Kubeflow SDK can help you manage Spark on Kubernetes ?

cc @yuchaoran2011 @vara-bonthu @nabuskey @ChenYi015 @jacobsalway @akshaychitneni @shravan-achar @lresende @bigsur0 @kubeflow/kubeflow-sdk-team @franciscojavierarceo

/remove-label lifecycle/needs-triage
/area spark

[CrashCumber]

Hello, @andreyvelich.
Thank you for your response.

My use-case is as follows:
I have a lot of spark jobs, which consist of a large python git repository with a lot of logic and class relationships, including inheritance. This production code includes a large number of data preparation jobs for different models.
Now, everything is being launched through the Airflow. And I'm researching the process of transferring all jobs to the Kubeflow Pipeline.

I already have a Kubernetes cluster with Kubeflow installed.
I studied the Spark Operator in detail and transferred the job to the Spark Application.
To run these jobs, I use the SparkApplication YAML manifest. The spark job code is included in the image, as well as the executable file.

apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
spec:
  type: Python
  mode: cluster
  image: "gitlab-registry/image_with_code"
  mainApplicationFile: "local:///opt/prepare_data.py"

I'm currently exploring ways to run and track jobs based on the Kubeflow Pipeline. At the moment, it seems that this can only be done through the Kubernetes python client, which is a relatively low-level approach. It seems to me that wrapping it in a convenient wrapper would be better solution.

In my opinion, this would be a python client/code that:

1. Create customized jobs based on the Spark image, configuration, resources, labels, and other annotations specified in the SparkApplication manifest.
2. Monitor the status of each job.
3. Provide access to the driver and executor logs.

In your message, it's similar to Option 1.
An example of pseudocode might look like this:

sparkBaseConfig = SparkBaseConfig(
    pod_security_context={"runAsUser": 185, "runAsNonRoot": True},
    security_context={...},
    annotations={...},
    image_pull_policy="Always",
    image_pull_secrets=[....],
    namespace="kubeflow",
    service_account="spark-service-account",
    spark_conf={"spark.driver.cores": "1",...},
    executor_resources={...},
    driver_resources={...},
    spark_version="3.2.4",
    restart_policy="Never",
    image="gitlab-registry/image_with_code",
    env={...},
    labels={...},
    ....
)

@dsl.component
def prepare_data():
    spark_application = SparkApplication(
        base_config=sparkBaseConfig,
        name="name1",
        spark_conf={"spark.hadoop.fs.defaultFS": "webhdfs://...."},  ## edit spark conf
        main_application_file="local:///opt/prepare_data1.py",
        env={...},  # add more env
    )

    spark_application.submit()

    while True:
        if datetime.now() > timeout_time:
            print(f"Timeout reached after {timeout_minutes} minutes waiting for Spark job completion")
            return False

        application_state = spark_application.get_status()
        print(application_state.get_logs())
        
        if application_state == "COMPLETED":
            return True
        elif application_state == "FAILED":
            return False

        time.sleep(10)

@dsl.component
def prepare_data2():
    spark_application = SparkApplication(
        base_config=sparkBaseConfig,
        name="name2",
        spark_conf={"spark.driver.cores": "2"},  ## overwrite some spark conf
        main_application_file="local:///opt/prepare_data2.py",
    )

    spark_application.submit()
    spark_application.wait_for_completion(timeout=10)  # replace where long operation


@dsl.pipeline
def pipeline():
    prepare_data()
    prepare_data2()

Perhaps there are other, more elegant ways to resolve this issue. I would appreciate hearing about them.

[andreyvelich]

Thanks for sharing this @CrashCumber!

Create customized jobs based on the Spark image, configuration, resources, labels, and other annotations specified in the SparkApplication manifest.
Monitor the status of each job.
Provide access to the driver and executor logs.

This is certainly possible, we do similar steps for TrainJob CRD.

I have question regarding the image and mainApplicationFile. Is this the only way to trigger Spark job in the Spark cluster started by SparkApplication?
If the answer is yes, users will be responsible to create Docker image by themselves (at least in the short-term until we automate it in the SDK, check this: #26).

As reference for KF Trainer, we wrap the users' code into container command, so it can be executed without building an image.

For example, PyTorch execution looks as follows:

apiVersion: trainer.kubeflow.org/v1alpha1
kind: TrainJob
metadata:
  name: m8b381169413
spec:
  trainer:
    command:
      - bash
      - -c
      - |2-
        if ! [ -x "$(command -v pip)" ]; then
            python -m ensurepip || python -m ensurepip --user || apt-get install python-pip
        fi

        PIP_DISABLE_PIP_VERSION_CHECK=1 python -m pip install --quiet --no-warn-script-location --index-url https://pypi.org/simple boto3

        read -r -d '' SCRIPT << EOM

        def train_fashion_mnist():
            import os
            print("Training Function!!")
        train_fashion_mnist()
        EOM
        printf "%s" "$SCRIPT" > "1666146569.py"
        torchrun "1666146569.py"
    numNodes: 3
    resourcesPerNode:
      limits:
        cpu: 3
        memory: 16Gi
        nvidia.com/gpu: 2
  runtimeRef:
    apiGroup: trainer.kubeflow.org
    kind: ClusterTrainingRuntime
    name: torch-distributed

[CrashCumber]

@andreyvelich, regarding your question about Spark - yes.
There are 3 ways to launch a Spark job on Kubernetes. In all cases, we will need an application file and an image.

1. Bash spark-submit
2. Python SparkSubmitOperator
3. Yaml SparkApplication

As for #26.
It's a good idea to create a job with a customized image for both Trainer and Spark.
I have an use-case where I needed to move the training phase of a model into a pipeline. This phase's code occupied a medium-sized repository.

However, I am not fully understanding how the customization of the image will work (from this #26). Will the target image be created based on the Dockerfile or some other method? What will this image be built on, in addition to the base image?

[andreyvelich]

However, I am not fully understanding how the customization of the image will work (from this #26). Will the target image be created based on the Dockerfile or some other method? What will this image be built on, in addition to the base image?

Correct, the rough idea is to dynamically build Docker image using the base_image and users's workspace. We can use Docker client API to automate image build + publishing.
Similar to how KFP build an image here for their containerize components(cc @kubeflow/wg-pipeline-leads )

[andreyvelich]

If you are following this issue, please check out this KEP to support SparkClient() in Kubeflow SDK: #163