Users can extend the backend, if they want to have any specific changes or different way to connect or submit spark job

For Trainer, backends represent various job submission (local subprocess, container, and Kubernetes). I am not sure if we can replicate it for Spark.

Here we have job submission using K8S Operator backend, Spark Connector backend , Gateway backend (not implemented - we just have abstract class).

I am wondering what is the main motivation to separate SessionClient() and BatchClient()?
Alternatively, we can just have unified SparkClient() which has sessions and batch APIs:

submit_job() <-- to create Spark Application and submit batch job
connect() <-- to create session and connect to existing Spark cluster

• BatchSparkClient users never see connect(), create_session() methods
• SparkSessionClient users never see submit_application(), wait_for_job_status() methods
• This prevents runtime errors: Can't call wait_for_job_status() on a session object.

This helps user to know which methods are available clearly :

 client = SparkClient()
  # User sees BOTH batch AND session methods
  client.submit_job(...)        # For batch
  client.connect(...)           # For session
  client.get_job(...)           # Works with connect() or batch()  ? 

 # event if we take arg in config: 
  client = SparkClient(mode="batch")
  client.create_session(...)  # IDE will not show error, but runtime error 

Do we know how the Spark session management will work with the new CRD: SparkCluster proposed by @ChenYi015 here: kubeflow/spark-operator#2744 ?

Added my view: kubeflow/spark-operator#2744 (comment)

If we want to create the spark cluster through the SDK then we need API like :

 from kubeflow.spark import SparkClusterManager

 manager = SparkClusterManager()

 # Create persistent cluster
 cluster = manager.create_cluster(
     name="prod",
     spark_version="4.0.1",
     init_workers=5,
     min_workers=2,
     max_workers=10,
     enable_hpa=True,
 )

# Cluster url to use for spark job submission: cluster.master_url

Can you explain reason of creating various backends? Can we just have an API: SparkClient().connect() which creates SparkConnect CR and connects to the existing cluster as we discussed?

The batch backend will have APIs like submit_application, wait_for_completion, get_logs, .. where user will just submit the job and can check the logs/results.
example: https://github.com/kubeflow/sdk/pull/158/files#diff-e692a5819ee6b1dc00cba3b58e91f058c0022d3ca9aa6f3ee468126f245eef89R89

But with interactive session spark client user will be able to run interactive SQL queries and DataFrame operations.
example: https://github.com/kubeflow/sdk/pull/158/files#diff-a5011f48c9d6d16ff6ddd65588f65a7c78abf5fbeb121cccb693c0892ce3a5aeR275

I am wondering if there is a way to allow users bypass function to SparkApplication similar to Trainer API: https://github.com/kubeflow/sdk?tab=readme-ov-file#run-your-first-pytorch-distributed-job

That might be interesting to explore how we can allow to submit SparkApplication without building an image.

Are we thinking an API like this ?

def my_job(input_path: str, output_path: str):                                                              
    from pyspark.sql import SparkSession
    spark = SparkSession.builder.getOrCreate()                                                         
    df = spark.read.csv(input_path)                                                                    
    df.groupBy("col").count().write.parquet(output_path)                                               
    spark.stop()                                                                                       
                                                                                                                    
client.submit_function(                                                                                
    func=my_job,                                                                                       
    func_args={"input_path": "s3://in", "output_path": "s3://out"}                                     
) 

Yeah, that might be interesting to explore.
We can also extend submit_job() API to accept main_application_file or function.

Let me add to KEP and will work on it.

Spark requires pre-allocated cluster resources (driver/executor specifications), so I could not find such way for running spark job. But we can explore in next version.

@vara-bonthu @ChenYi015 @nabuskey Do you have any ideas on how we can dynamically inject the users' code into Spark driver without re-building an image or put Python file into Pod's filesystem?

I have one idea if we have can leverage PodTemplate spec.
Let's say we define InitContainer that simply creates file in the Pod's filesystem with the Spark code in the pre-defined location (e.g. local:///opt/spark/spark_job.py). This initContainer can share emptyDir with the driver container, so driver container can see the file. To create the file, we can use the same script as for Trainer:

container:
  command:
    - bash
    - -c
    - |
      read -r -d '' SCRIPT << EOM
      def my_job(input_path: str, output_path: str):                                                           
          from pyspark.sql import SparkSession
          spark = SparkSession.builder.getOrCreate()                                                         
          df = spark.read.csv(input_path)                                                                    
          df.groupBy("col").count().write.parquet(output_path)                                               
          spark.stop()    

      EOM
      printf "%s" "$SCRIPT" > "/opt/spark/spark_job.py"

WDYT ?

cc @shravan-achar @akshaychitneni @bigsur0 for feedback on this.

@andreyvelich

Sorry to interrupt. I implemented a decorator that uses inspect to get the function's body and form an executable file. It also parses the function's signature into submit parameters (they must be JSON-serializable). I could implement this if you're interested.

Example

@spark_submit
def my_task(first, second):
    ...

@aagumin Yeah, this looks similar to how we do this in TrainerClient(): https://github.com/kubeflow/sdk/blob/main/kubeflow/trainer/backends/kubernetes/utils.py#L297-L301
I think, having the consistent UX across Data Processing (e.g. SparkJobs) and Training (e.g. TrainJob) workloads would be really cool.
As long as SparkApplication CRD can support initContainer via PodTemplate, that should be relatively easy to implement.

cc @franciscojavierarceo @kubeflow/kubeflow-sdk-team

+1

This is pretty interesting experience for dev -> prod Spark lifecycle.
cc @shravan-achar @akshaychitneni @bigsur0 to explore.

cc @kubeflow/wg-pipeline-leads to explore

How will this be better than a native SparkConnect call? I can see that you can’t replace a Connect client with a regular one without changing interfaces, considering that the options for configuring a Connect session are significantly more limited.

We will have more kubeflow compatible APIs and infra.

If a PR is added for gRPC submit support and the ability to implement a custom submit mechanism, then this becomes relevant.

I like the idea of builder(), but shall this comes in the 2nd phase?
We have similar ideas with @kubeflow/kubeflow-sdk-team for TrainerClient and OptimizerClient, so maybe we can unify the experience?

Initially, we can just allow users to specify minimal number of parameters in the connect() API.

But I am happy to debate it during our next SDK call on Dec 17th: https://docs.google.com/document/d/1jH2WAX2ePxOfI4JuiVK9nPlesDMiyg67xzLwhpR7wTQ/edit?tab=t.0#heading=h.jarjwwjkqmul

Agree, I'd like us to start with constructor and config objects aligning with existing clients and later expand it with builder approach

Will SparkApplication and SparkConnect CRD always have the same APIs? I am wondering if we can re-use the same builder for them.
cc @ChenYi015

This would be internal changes, we can take care of it later as well.

I think, namespace should be part of KubernetesBackendConfig in SparkClient to align with other clients: https://github.com/kubeflow/sdk/blob/main/kubeflow/trainer/backends/kubernetes/backend.py#L42-L43

We should whether we need to move it under: kubeflow.pipelines.components.
We can also leverage KFP lightweight Python tasks to submit such pipelines.

cc @franciscojavierarceo @kubeflow/wg-pipeline-leads

Will this integrate well with our current multi-tenancy architecture ? https://github.com/kubeflow/manifests#architecture

So if i execute this from a jupyterlab will it just use the current namespace?

Yes

I am missing a bit the CPU and memory requests vs Limit . I only see

.driver(cores=2, memory="4g")
.executor(cores=4, memory="8g")

which is somehow mixing up requests with limits. Furthermore we should make sure that it also works with the integrated seaweedfs S3.

Thanks for pointing it out. There are lot of conf is spark so anyways we need wider key value pair set.

I guess, we can remove this from the version 1, right?
And implement this in the 2nd phase if needed.

Yes +1

Can you move Builder to the future plans, since we will implement it later if needed.

Yeah, we will start simpler in first MVP

Does SparkConnect CRD support ssl and auth?

No built-in SSL/TLS or authentication support in the SparkConnect CRD. I was thinking for external servers or submitting spark job. But let me remove it for now - we will think later on this.

Removed it for now a2a4837

We don't need this, right ?

If user want to provide specific name - let's park it for phase 2

WIll name be equal to the SparkConnect CR name? Can we move it to options for now, to be consistent with Trainer.

Let's say that job submission APIs will be implemented in the phase 2.