feat: [KEP-107] Support spark_conf for Spark batch job submission - #697
feat: [KEP-107] Support spark_conf for Spark batch job submission#697Goku2099 wants to merge 2 commits into
Conversation
Signed-off-by: Sameer_yadav <159073326+Goku2099@users.noreply.github.com>
|
[APPROVALNOTIFIER] This PR is NOT APPROVED This pull-request has been approved by: The full list of commands accepted by this bot can be found here. DetailsNeeds approval from an approver in each of these files:Approvers can indicate their approval by writing |
Signed-off-by: Sameer_yadav <159073326+Goku2099@users.noreply.github.com>
tariq-hasan
left a comment
tariq-hasan
left a comment
There was a problem hiding this comment.
Hi @Goku2099! I have added a few comments.
|
|
||
| print("test execution complete") | ||
|
|
||
|
|
There was a problem hiding this comment.
For the cases of test_get_spark_application_cr_from_file_job and test_get_spark_application_cr_from_func_job I think probably a good idea to have dedicated test cases per behavior (spark_conf included) similar to what we have for test_build_spark_connect_cr.
| num_executors=None, | ||
| resources_per_executor=None, | ||
| spark_conf=None, | ||
| ) |
There was a problem hiding this comment.
I think better to combine test_submit_job_validation and test_submit_job_success and have all test cases grouped under test_submit_job for consistency with how unit tests in the packages are structured in general - using the TestCase pattern.
| arguments=arguments or None, | ||
| driver=get_spark_job_driver_spec(), | ||
| executor=get_spark_job_executor_spec( | ||
| num_executors=num_executors, | ||
| resources_per_executor=resources_per_executor, | ||
| ), | ||
| spark_conf=spark_conf, |
There was a problem hiding this comment.
We can do spark_conf=spark_conf or None, to match arguments=arguments or None,.
Same in get_spark_application_cr_from_func_job
| ) | ||
|
|
||
| return instances, cores, memory | ||
|
|
There was a problem hiding this comment.
We need to have a precedence rule here for the case when executor resources are passed in via spark_conf.
In the case when executor resources are passed in via spark_conf they should have precedence over any other resource specification.
Granted - a user would not necessarily want to use spark_conf to pass in executor resources. They'd want to use spark_conf for advanced settings. But either way this is a correctness issue so it would need to be addressed.
| ( | ||
| FileJob(file_source="s3://bucket/job.py"), | ||
| {"spark.executor.memory": "4g"}, | ||
| None, | ||
| None, | ||
| NotImplementedError, | ||
| ), |
There was a problem hiding this comment.
Can we add a test case to have coverage for spark_conf?
| else: | ||
| raise ValueError(f"Invalid CPU type '{type(cpu)}'. Expected str or int.") |
There was a problem hiding this comment.
Similar to how we validate cpu do we also want to validate the type for spark_conf key and value?
| spark_conf=spark_conf, | ||
| ) | ||
|
|
||
| try: |
There was a problem hiding this comment.
Having spark_conf test cases for the backend would be nice as well.
2 participants
Summary
This PR adds support for
spark_confwhen submitting Spark batch jobs withSparkClient.Previously, passing
spark_conftoSparkClient.submit_job()resulted in aNotImplementedError. With this change, user-provided Spark configuration is propagated to the generatedSparkApplicationresource.Changes
NotImplementedErrorforspark_confinSparkClient.submit_job().spark_confthrough the Kubernetes backend.SparkApplication.spec.sparkConfwith the provided Spark configuration.spark_confpropagation to the generatedSparkApplication.Example
closes: #646
ref: #520