diff --git a/internal/pkg/object/command/sparkeks/entrypoint.go b/internal/pkg/object/command/sparkeks/entrypoint.go new file mode 100644 index 00000000..d708e11a --- /dev/null +++ b/internal/pkg/object/command/sparkeks/entrypoint.go @@ -0,0 +1,134 @@ +package sparkeks + +import ( + "path" + "strings" + + "github.com/kubeflow/spark-operator/v2/api/v1beta2" +) + +// --- JAR application type (JVM language) resolution --- +// Add an entry to support another JVM application type. +//e.g. Combination: applicationType (Java) -> EntrypointStrategy (JAR) + (Class name) + +var jarApplicationTypes = map[string]v1beta2.SparkApplicationType{ + "scala": v1beta2.SparkApplicationTypeScala, + "java": v1beta2.SparkApplicationTypeJava, +} + +const defaultJarApplicationType = v1beta2.SparkApplicationTypeScala + +// resolveJarApplicationType maps an optional configured application type (e.g. "Java", "Scala", +// case-insensitive) to its SparkApplication type, falling back to the default (Scala) when +// unset or unrecognized. +func resolveJarApplicationType(applicationType string) v1beta2.SparkApplicationType { + if t, ok := jarApplicationTypes[strings.ToLower(strings.TrimSpace(applicationType))]; ok { + return t + } + return defaultJarApplicationType +} + +type entrypointStrategy interface { + apply(spec *v1beta2.SparkApplicationSpec) error +} + +func buildArguments(override []string, appName, queryURI, user, resultURI string, returnResult bool) []string { + if len(override) > 0 { + return override + } + args := []string{appName, queryURI, user} + if returnResult { + args = append(args, resultURI) + } + return args +} + +type jarEntrypointStrategy struct { + appType v1beta2.SparkApplicationType + mainClass string + appName string + queryURI string + user string + resultURI string + returnResult bool + arguments []string +} + +func (s jarEntrypointStrategy) apply(spec *v1beta2.SparkApplicationSpec) error { + mainClass := strings.TrimSpace(s.mainClass) + if mainClass == "" { + return ErrMissingEntryPoint + } + spec.Type = s.appType + spec.MainClass = &mainClass + + spec.Arguments = buildArguments(s.arguments, s.appName, s.queryURI, s.user, s.resultURI, s.returnResult) + return nil +} + +type sqlWrapperEntrypointStrategy struct { + appName string + queryURI string + user string + resultURI string + returnResult bool + arguments []string +} + +func (s sqlWrapperEntrypointStrategy) apply(spec *v1beta2.SparkApplicationSpec) error { + spec.Arguments = buildArguments(s.arguments, s.appName, s.queryURI, s.user, s.resultURI, s.returnResult) + return nil +} + +// entrypointFactory builds the entrypoint strategy for a job from its execution context. +type entrypointFactory func(execCtx *executionContext) entrypointStrategy + +var entrypointStrategiesByExt = map[string]entrypointFactory{ + ".jar": newJarEntrypointStrategy, + ".py": newSQLWrapperEntrypointStrategy, +} + +var defaultEntrypointFactory entrypointFactory = newSQLWrapperEntrypointStrategy + +func newEntrypointStrategy(execCtx *executionContext) entrypointStrategy { + ext := strings.ToLower(path.Ext(execCtx.commandContext.WrapperURI)) + factory, ok := entrypointStrategiesByExt[ext] + if !ok { + factory = defaultEntrypointFactory + } + return factory(execCtx) +} + +func newJarEntrypointStrategy(execCtx *executionContext) entrypointStrategy { + jobContext := execCtx.jobContext + + var mainClass, applicationType string + if jobContext.Parameters != nil { + mainClass = jobContext.Parameters.EntryPoint + applicationType = jobContext.Parameters.ApplicationType + } + + return jarEntrypointStrategy{ + appType: resolveJarApplicationType(applicationType), + mainClass: mainClass, + appName: execCtx.appName, + queryURI: execCtx.s3aQueryURI, + user: execCtx.job.User, + resultURI: execCtx.s3aResultURI, + returnResult: jobContext.ReturnResult, + arguments: jobContext.Arguments, + } +} + +func newSQLWrapperEntrypointStrategy(execCtx *executionContext) entrypointStrategy { + jobContext := execCtx.jobContext + + return sqlWrapperEntrypointStrategy{ + appName: execCtx.appName, + queryURI: execCtx.s3aQueryURI, + user: execCtx.job.User, + resultURI: execCtx.s3aResultURI, + returnResult: jobContext.ReturnResult, + arguments: jobContext.Arguments, + } +} diff --git a/internal/pkg/object/command/sparkeks/entrypoint_test.go b/internal/pkg/object/command/sparkeks/entrypoint_test.go new file mode 100644 index 00000000..7a24f19d --- /dev/null +++ b/internal/pkg/object/command/sparkeks/entrypoint_test.go @@ -0,0 +1,154 @@ +package sparkeks + +import ( + "reflect" + "testing" + + "github.com/kubeflow/spark-operator/v2/api/v1beta2" + + "github.com/patterninc/heimdall/pkg/object" + "github.com/patterninc/heimdall/pkg/object/job" +) + +func TestResolveJarApplicationType(t *testing.T) { + tests := []struct { + name string + applicationType string + expected v1beta2.SparkApplicationType + }{ + {name: "scala lowercase", applicationType: "scala", expected: v1beta2.SparkApplicationTypeScala}, + {name: "java lowercase", applicationType: "java", expected: v1beta2.SparkApplicationTypeJava}, + {name: "java mixed case", applicationType: "Java", expected: v1beta2.SparkApplicationTypeJava}, + {name: "scala uppercase", applicationType: "SCALA", expected: v1beta2.SparkApplicationTypeScala}, + {name: "java with surrounding whitespace", applicationType: " java ", expected: v1beta2.SparkApplicationTypeJava}, + {name: "empty defaults to scala", applicationType: "", expected: defaultJarApplicationType}, + {name: "whitespace only defaults to scala", applicationType: " ", expected: defaultJarApplicationType}, + {name: "unrecognized defaults to scala", applicationType: "python", expected: defaultJarApplicationType}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := resolveJarApplicationType(tt.applicationType); got != tt.expected { + t.Errorf("resolveJarApplicationType(%q) = %v, want %v", tt.applicationType, got, tt.expected) + } + }) + } +} +func TestNewEntrypointStrategy_Selection(t *testing.T) { + tests := []struct { + name string + wrapperURI string + expectedType reflect.Type + }{ + {name: "jar selects jar strategy", wrapperURI: "s3://bucket/app.jar", expectedType: reflect.TypeOf(jarEntrypointStrategy{})}, + {name: "py selects sql wrapper strategy", wrapperURI: "s3://bucket/wrapper.py", expectedType: reflect.TypeOf(sqlWrapperEntrypointStrategy{})}, + {name: "jar uppercase extension", wrapperURI: "s3://bucket/APP.JAR", expectedType: reflect.TypeOf(jarEntrypointStrategy{})}, + {name: "py mixed case selects sql strategy", wrapperURI: "s3://bucket/Wrapper.Py", expectedType: reflect.TypeOf(sqlWrapperEntrypointStrategy{})}, + {name: "unknown extension defaults to sql wrapper", wrapperURI: "s3://bucket/app.sh", expectedType: reflect.TypeOf(sqlWrapperEntrypointStrategy{})}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + execCtx := newTestExecutionContext(tt.wrapperURI, &jobContext{}, "alice") + strategy := newEntrypointStrategy(execCtx) + if got := reflect.TypeOf(strategy); got != tt.expectedType { + t.Errorf("newEntrypointStrategy(%q) type = %v, want %v", tt.wrapperURI, got, tt.expectedType) + } + }) + } +} + +func TestNewJarEntrypointStrategy_Apply(t *testing.T) { + jobCtx := &jobContext{ + ReturnResult: true, + Parameters: &jobParameters{ + EntryPoint: "com.example.Main", + ApplicationType: "java", + }, + } + execCtx := newTestExecutionContext("s3://bucket/app.jar", jobCtx, "alice") + + strategy := newEntrypointStrategy(execCtx) + spec := &v1beta2.SparkApplicationSpec{} + if err := strategy.apply(spec); err != nil { + t.Fatalf("apply() returned unexpected error: %v", err) + } + + if spec.Type != v1beta2.SparkApplicationTypeJava { + t.Errorf("spec.Type = %v, want %v", spec.Type, v1beta2.SparkApplicationTypeJava) + } + if spec.MainClass == nil || *spec.MainClass != "com.example.Main" { + t.Errorf("spec.MainClass = %v, want com.example.Main", spec.MainClass) + } + expectedArgs := []string{"spark-sql-job-test", "s3a://bucket/query.sql", "alice", "s3a://bucket/result"} + if !reflect.DeepEqual(spec.Arguments, expectedArgs) { + t.Errorf("spec.Arguments = %v, want %v", spec.Arguments, expectedArgs) + } +} + +func TestNewSQLWrapperEntrypointStrategy_Apply(t *testing.T) { + jobCtx := &jobContext{ReturnResult: false} + execCtx := newTestExecutionContext("s3://bucket/wrapper.py", jobCtx, "bob") + + strategy := newEntrypointStrategy(execCtx) + spec := &v1beta2.SparkApplicationSpec{} + if err := strategy.apply(spec); err != nil { + t.Fatalf("apply() returned unexpected error: %v", err) + } + + if spec.MainClass != nil { + t.Errorf("spec.MainClass = %v, want nil for sql wrapper", spec.MainClass) + } + expectedArgs := []string{"spark-sql-job-test", "s3a://bucket/query.sql", "bob"} + if !reflect.DeepEqual(spec.Arguments, expectedArgs) { + t.Errorf("spec.Arguments = %v, want %v", spec.Arguments, expectedArgs) + } +} + +func TestJarEntrypointStrategy_Apply_MissingEntryPoint(t *testing.T) { + tests := []struct { + name string + jobCtx *jobContext + }{ + {name: "nil parameters", jobCtx: &jobContext{}}, + {name: "empty entry_point", jobCtx: &jobContext{Parameters: &jobParameters{EntryPoint: ""}}}, + {name: "whitespace entry_point", jobCtx: &jobContext{Parameters: &jobParameters{EntryPoint: " "}}}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + execCtx := newTestExecutionContext("s3://bucket/app.jar", tt.jobCtx, "alice") + strategy := newEntrypointStrategy(execCtx) + spec := &v1beta2.SparkApplicationSpec{} + err := strategy.apply(spec) + if err != ErrMissingEntryPoint { + t.Errorf("apply() error = %v, want %v", err, ErrMissingEntryPoint) + } + if spec.MainClass != nil { + t.Errorf("spec.MainClass = %v, want nil when entry_point is missing", spec.MainClass) + } + }) + } +} + +func TestSQLWrapperEntrypointStrategy_Apply_NoEntryPointRequired(t *testing.T) { + execCtx := newTestExecutionContext("s3://bucket/wrapper.py", &jobContext{}, "alice") + strategy := newEntrypointStrategy(execCtx) + spec := &v1beta2.SparkApplicationSpec{} + if err := strategy.apply(spec); err != nil { + t.Errorf("apply() returned unexpected error: %v", err) + } +} + +func newTestExecutionContext(wrapperURI string, jobCtx *jobContext, user string) *executionContext { + return &executionContext{ + job: &job.Job{ + Object: object.Object{User: user}, + }, + commandContext: &commandContext{WrapperURI: wrapperURI}, + jobContext: jobCtx, + appName: "spark-sql-job-test", + s3aQueryURI: "s3a://bucket/query.sql", + s3aResultURI: "s3a://bucket/result", + } +} diff --git a/internal/pkg/object/command/sparkeks/sparkeks.go b/internal/pkg/object/command/sparkeks/sparkeks.go index 909561ef..89c48871 100644 --- a/internal/pkg/object/command/sparkeks/sparkeks.go +++ b/internal/pkg/object/command/sparkeks/sparkeks.go @@ -52,6 +52,8 @@ const ( defaultSparkAppAPIVersion = "sparkoperator.k8s.io/v1beta2" defaultSparkAppKind = "SparkApplication" + defaultApplicationType = v1beta2.SparkApplicationTypePython + queriesPath = "queries" resultsPath = "results" logsPath = "logs" @@ -92,6 +94,7 @@ var ( ErrKubeConfig = fmt.Errorf("failed to configure Kubernetes client: ensure EKS cluster access is properly configured") ErrApplicationSpec = fmt.Errorf("failed to load or parse SparkApplication template") ErrSparkApplicationFile = fmt.Errorf("failed to read SparkApplication application template file: check file path and permissions") + ErrMissingEntryPoint = fmt.Errorf("entry_point is required for .jar wrapper_uri: set parameters.entry_point to the fully-qualified main class") ) type commandContext struct { @@ -104,10 +107,13 @@ type commandContext struct { type jobParameters struct { Properties map[string]string `yaml:"properties,omitempty" json:"properties,omitempty"` + EntryPoint string `yaml:"entry_point,omitempty" json:"entry_point,omitempty"` + ApplicationType string `yaml:"application_type,omitempty" json:"application_type,omitempty"` } type jobContext struct { Query string `yaml:"query,omitempty" json:"query,omitempty"` + Arguments []string `yaml:"arguments,omitempty" json:"arguments,omitempty"` Parameters *jobParameters `yaml:"parameters,omitempty" json:"parameters,omitempty"` ReturnResult bool `yaml:"return_result,omitempty" json:"return_result,omitempty"` } @@ -139,6 +145,8 @@ type executionContext struct { appName string queryURI string resultURI string + s3aQueryURI string + s3aResultURI string logURI string sparkApp *v1beta2.SparkApplication submittedApp *v1beta2.SparkApplication @@ -370,6 +378,8 @@ func buildExecutionContextAndURI(ctx context.Context, r *plugin.Runtime, j *job. execCtx.appName = fmt.Sprintf("%s-%s", applicationPrefix, j.ID) execCtx.queryURI = fmt.Sprintf("%s/%s/%s/%s", s.JobsURI, j.ID, queriesPath, queryFileName) execCtx.resultURI = fmt.Sprintf("%s/%s/%s", s.JobsURI, j.ID, resultsPath) + execCtx.s3aQueryURI = updateS3ToS3aURI(execCtx.queryURI) + execCtx.s3aResultURI = updateS3ToS3aURI(execCtx.resultURI) execCtx.logURI = fmt.Sprintf("%s/%s/%s", s.JobsURI, j.ID, logsPath) // Upload query to S3 @@ -472,6 +482,7 @@ func updateS3ToS3aURI(uri string) string { return strings.ReplaceAll(uri, s3Prefix, s3aPrefix) } + // getS3FileURI finds a file in an S3 directory that matches the given extension. func getS3FileURI(ctx context.Context, awsConfig aws.Config, directoryURI, matchingExtension string) (string, error) { s3Parts := rxS3.FindAllStringSubmatch(directoryURI, -1) @@ -718,7 +729,7 @@ func updateKubeConfig(ctx context.Context, execCtx *executionContext) (string, e } // applySparkOperatorConfig consolidates all Spark Operator configuration updates and overrides. -func applySparkOperatorConfig(execCtx *executionContext) { +func applySparkOperatorConfig(execCtx *executionContext) error { sparkApp := execCtx.sparkApp jobContext := execCtx.jobContext clusterContext := execCtx.clusterContext @@ -730,15 +741,11 @@ func applySparkOperatorConfig(execCtx *executionContext) { // Set main application file and arguments if execCtx.commandContext.WrapperURI != "" { s3aWrapperURI := updateS3ToS3aURI(execCtx.commandContext.WrapperURI) - s3aQueryURI := updateS3ToS3aURI(execCtx.queryURI) - s3aResultURI := updateS3ToS3aURI(execCtx.resultURI) - mainAppFile := s3aWrapperURI - if jobContext.ReturnResult { - sparkApp.Spec.Arguments = []string{execCtx.appName, s3aQueryURI, execCtx.job.User, s3aResultURI} - } else { - sparkApp.Spec.Arguments = []string{execCtx.appName, s3aQueryURI, execCtx.job.User} + sparkApp.Spec.MainApplicationFile = &s3aWrapperURI + // Type / MainClass / Arguments are set by the entrypoint strategy. + if err := newEntrypointStrategy(execCtx).apply(&sparkApp.Spec); err != nil { + return err } - sparkApp.Spec.MainApplicationFile = &mainAppFile } if sparkApp.Spec.SparkConf == nil { @@ -844,6 +851,12 @@ func applySparkOperatorConfig(execCtx *executionContext) { for k, v := range jobContext.Parameters.Properties { sparkApp.Spec.SparkConf[k] = v } + + if sparkApp.Spec.Type == "" { + sparkApp.Spec.Type = defaultApplicationType + } + + return nil } // generateSparkApp generates the Spark application from the template. @@ -856,7 +869,9 @@ func generateSparkApp(execCtx *executionContext) error { execCtx.sparkApp = sparkApp // Apply all configurations and overrides from contexts - applySparkOperatorConfig(execCtx) + if err := applySparkOperatorConfig(execCtx); err != nil { + return err + } return nil } @@ -967,7 +982,12 @@ func (e *executionContext) monitorJobAndCollectLogs(ctx context.Context) error { if state == v1beta2.ApplicationStateUnknown { msg = sparkAppUnknownStateMsg } - return fmt.Errorf("%s", msg) + errorMessage := sparkApp.Status.AppState.ErrorMessage + if errorMessage == "" { + errorMessage = unknownErrorMsg + } + e.runtime.Stderr.WriteString(fmt.Sprintf("%s: %s\n", msg, errorMessage)) + return fmt.Errorf("%s: %s", msg, errorMessage) } } } diff --git a/plugins/sparkeks/README.md b/plugins/sparkeks/README.md index 0c923947..7a66956b 100644 --- a/plugins/sparkeks/README.md +++ b/plugins/sparkeks/README.md @@ -10,6 +10,7 @@ The Spark EKS plugin enables submitting Spark jobs to an AWS EKS cluster using t - Supports custom SparkApplication YAML templates - IAM role assumption for cross-account EKS access - Configurable Spark job resources and properties +- **JAR / `--class` entrypoints** — run a JVM (Scala/Java) application by main class, alongside the default Python SQL wrapper ## Configuration @@ -59,6 +60,90 @@ The Spark EKS plugin enables submitting Spark jobs to an AWS EKS cluster using t } ``` +## Entrypoints: SQL wrapper vs JAR + +The plugin picks an entrypoint from the **file extension of the command's `wrapper_uri`**. The wrapper +file itself is always submitted as `MainApplicationFile`; only `Type`, `MainClass`, and `Arguments` +differ between the two. + +| `wrapper_uri` ends in | Entrypoint | `Spec.Type` | `Spec.MainClass` | +|---|---|---|---| +| `.py` | Python SQL wrapper (default, pre-existing behavior) | `Python` | not set | +| `.jar` | JVM JAR, run by main class | `Scala` or `Java` | `entry_point` | +| anything else | falls back to the SQL wrapper | `Python` | not set | + +### JAR configuration + +Set on the **job context** under `parameters`: + +| Field | Required | Description | +|---|---|---| +| `entry_point` | yes (for JAR) | Fully-qualified main class, e.g. `com.org.customapplication.main`. Becomes `--class`. | +| `application_type` | no | `Scala` or `Java`, case-insensitive. Defaults to `Scala` when empty or unrecognized. | + +```json +{ + "query": "SELECT key, value FROM my_table", + "wrapper_uri": "s3://mybucket/application-assembly.jar", + "parameters": { + "entry_point": "com.org.customapplication.main", + "application_type": "Scala", + "properties": { + //sparkeks properties + } + }, + "return_result": false +} +``` + +### Arguments + +If the job context sets `arguments`, that list is passed to the app **verbatim** and the managed +list is not used — letting callers match an app's exact CLI contract (e.g. an app that expects +`[app_name, user, query, s3_path]`). Jobs that set no `arguments` keep the managed behavior, so +existing contracts are unchanged. + + +#### Example 1 — context provided + +No `arguments`; the SQL travels as an uploaded `queryURI`: + +```json +{ + "context": { + "query": "SELECT * FROM my_table LIMIT 10;", + "return_result": true + } +} +``` + +The app receives `[appName, query, user, resultURI]`. + +#### Example 2 — verbatim override + +`arguments` supplies the app's exact CLI shape (here `[app_name, user, query, s3_path]`); the +managed list is ignored: + +```json +{ + "context": { + "arguments": [ + "my-app", + "some_user", + "SELECT key, value FROM my_table", + "s3://mybucket/output/v1" + ], + "parameters": { + "entry_point": "com.org.customapplication.main", + "application_type": "Scala" + } + } +} +``` + +Here the app receives exactly those four strings. The SQL is passed inline, so mind arg-length +limits for very large queries. + ## Usage Submit a job using the API: