This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 0562a9047b [#13703] feat(k8s-job-executor): Apply spark_conf of a
built-in job when submitting it (#13982)
0562a9047b is described below
commit 0562a9047b14733862af9550474991b3593b5184
Author: Jerry Shao <[email protected]>
AuthorDate: Mon Oct 26 19:55:04 2026 +0800
[#13703] feat(k8s-job-executor): Apply spark_conf of a built-in job when
submitting it (#13982)
### What changes were proposed in this pull request?
- k8s job executor: for a `builtin-` template, parse its `--spark-conf`
argument (`spark_conf`) and merge it into the SparkApplication
`spec.sparkConf` (server `spark.conf.*` defaults < environments <
template `configs` < `spark_conf`). In `spark_conf`, `spark.master` /
`spark.submit.deployMode` and all `spark.kubernetes.*` keys are ignored
with a warning; an invalid `spark_conf` rejects the run.
- k8s job executor: reject a `spark.master` that isn't a `k8s://` master
(from the template or the executor defaults), like a client deploy mode,
instead of silently dropping it.
- Blank Spark configuration values mean "unset": the local job executor
doesn't pass them to spark-submit, and the k8s job executor doesn't let
them override its defaults.
- Built-in templates: `spark_master` defaults to empty instead of
`local[*]` (spark-submit's own default), template versions bumped.
- Built-in jobs (driver): ignore `spark.master` /
`spark.submit.deployMode` from `spark_conf` once spark-submit has set
them, drop `spark.kubernetes.*` and blank values.
- Optimizer `submit-update-stats-job`: no longer defaults `spark_master`
to `local[*]` or takes it from `spark.master` in `spark_conf`.
- Docs and the "Missing Iceberg Spark session extensions" error message
no longer recommend `spark.jars` in `spark_conf` for the local job
executor.
### Why are the changes needed?
`spark_conf` of a built-in job is applied by the driver after it starts,
too late for submit-time configurations: `spark.jars` /
`spark.jars.packages` don't reach the driver classpath, and
`spark.master` in `spark_conf` made a job run in local mode inside the
driver pod while reporting success. Built-in templates also defaulted to
`local[*]`, which only worked on Kubernetes because the executor
silently dropped the master.
Fix: #13703
### Does this PR introduce _any_ user-facing change?
- With the k8s job executor, `spark_conf` of a built-in job takes effect
at submit time (e.g. `spark.jars`, `spark.jars.packages`,
`spark.driver.*`); its `spark.kubernetes.*` keys are ignored.
- A job whose `spark.master` isn't a Kubernetes master is rejected by
the k8s job executor; an invalid `spark_conf` of a built-in job is
rejected.
- Built-in template versions: `v3`
(`builtin-iceberg-update-manifest-stats`: `v2`); `spark_master` defaults
to empty.
- No new configuration keys.
### How was this patch tested?
- New unit tests in `TestSparkApplicationUtils`,
`TestSparkProcessBuilder`, `TestIcebergJobUtils`, `TestOptimizerCmd` and
the template tests.
- Manually on OrbStack (spark-kubernetes-operator 0.9.0, Spark 3.5.9,
image without the Iceberg runtime): built-in Iceberg jobs with the
runtime from `spark.jars` (https) and `spark.jars.packages` in
`spark_conf` succeed; `spark.master` and `spark.kubernetes.*` in
`spark_conf` are ignored; a local master and an invalid `spark_conf` are
rejected; an empty `spark_conf` keeps the executor's default
`spark.jars`.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
Co-authored-by: Qi Yu <[email protected]>
---
.../gravitino/job/local/SparkProcessBuilder.java | 7 +-
.../job/local/TestSparkProcessBuilder.java | 24 ++
.../optimizer-cli-reference.md | 37 ++-
.../optimizer-configuration.md | 12 +-
.../optimizer-troubleshooting.md | 15 +-
docs/table-maintenance-service/optimizer.md | 9 +-
.../jobs/iceberg/IcebergExpireSnapshotsJob.java | 2 +-
.../maintenance/jobs/iceberg/IcebergJobUtils.java | 96 ++++++-
.../jobs/iceberg/IcebergRemoveOrphanFilesJob.java | 2 +-
.../jobs/iceberg/IcebergRewriteDataFilesJob.java | 2 +-
.../jobs/iceberg/IcebergRewriteManifestsJob.java | 2 +-
.../iceberg/IcebergUpdateManifestStatsJob.java | 4 +-
.../iceberg/IcebergUpdateStatsAndMetricsJob.java | 4 +-
.../maintenance/jobs/spark/SparkPiJob.java | 6 +-
.../iceberg/TestIcebergExpireSnapshotsJob.java | 2 +-
.../jobs/iceberg/TestIcebergJobUtils.java | 43 ++++
.../iceberg/TestIcebergRewriteDataFilesJob.java | 2 +-
.../iceberg/TestIcebergRewriteManifestsJob.java | 2 +-
.../iceberg/TestIcebergUpdateManifestStatsJob.java | 2 +-
.../jobs/iceberg/TestIcebergUpdateStatsJob.java | 2 +-
.../maintenance/jobs/spark/TestSparkPiJob.java | 4 +-
.../common/util/IcebergSparkConfigUtils.java | 4 +-
.../common/util/TestIcebergSparkConfigUtils.java | 7 +
.../command/SubmitUpdateStatsJobCommand.java | 22 +-
.../maintenance/optimizer/TestOptimizerCmd.java | 17 +-
plugins/k8s-job-executor/build.gradle.kts | 3 +
.../job/k8s/spark/SparkApplicationUtils.java | 164 +++++++++++-
.../job/k8s/TestK8sJobExecutorShadedJar.java | 3 +-
.../job/k8s/spark/TestSparkApplicationUtils.java | 279 ++++++++++++++++++++-
29 files changed, 717 insertions(+), 61 deletions(-)
diff --git
a/core/src/main/java/org/apache/gravitino/job/local/SparkProcessBuilder.java
b/core/src/main/java/org/apache/gravitino/job/local/SparkProcessBuilder.java
index c789b26ed0..ec79f5e4b7 100644
--- a/core/src/main/java/org/apache/gravitino/job/local/SparkProcessBuilder.java
+++ b/core/src/main/java/org/apache/gravitino/job/local/SparkProcessBuilder.java
@@ -117,8 +117,13 @@ public class SparkProcessBuilder extends
LocalProcessBuilder {
commandList.add(String.join(",", sparkJobTemplate.archives()));
}
- // Add the Spark configs
+ // Add the Spark configs. A blank value means the configuration isn't set,
as it is what an
+ // optional template parameter such as {{spark_jars:-}} becomes, and
spark-submit would reject
+ // or misread a blank value of some configurations, such as spark.master.
for (Map.Entry<String, String> entry :
sparkJobTemplate.configs().entrySet()) {
+ if (StringUtils.isBlank(entry.getValue())) {
+ continue;
+ }
commandList.add("--conf");
commandList.add(entry.getKey() + "=" + entry.getValue());
}
diff --git
a/core/src/test/java/org/apache/gravitino/job/local/TestSparkProcessBuilder.java
b/core/src/test/java/org/apache/gravitino/job/local/TestSparkProcessBuilder.java
index e3d8f4295d..c3612a7edc 100644
---
a/core/src/test/java/org/apache/gravitino/job/local/TestSparkProcessBuilder.java
+++
b/core/src/test/java/org/apache/gravitino/job/local/TestSparkProcessBuilder.java
@@ -150,6 +150,30 @@ public class TestSparkProcessBuilder {
Assertions.assertTrue(command4.contains("arg2"));
}
+ @Test
+ public void testGenerateSparkSubmitCommandSkipsBlankConfigs() {
+ SparkJobTemplate template =
+ SparkJobTemplate.builder()
+ .withName("template")
+ .withExecutable("/path/to/spark-demo.jar")
+ .withClassName("com.example.MainClass")
+ .withConfigs(
+ ImmutableMap.of(
+ "spark.master", "",
+ "spark.jars", " ",
+ "spark.executor.memory", "2g"))
+ .build();
+
+ List<String> command =
+
SparkProcessBuilder.generateSparkSubmitCommand("/path/to/spark-submit",
template);
+
+ Assertions.assertEquals(1, Collections.frequency(command, "--conf"));
+ int idxConf = command.indexOf("--conf");
+ Assertions.assertEquals("spark.executor.memory=2g", command.get(idxConf +
1));
+ Assertions.assertTrue(command.stream().noneMatch(arg ->
arg.startsWith("spark.master")));
+ Assertions.assertTrue(command.stream().noneMatch(arg ->
arg.startsWith("spark.jars")));
+ }
+
@Test
public void testResolveSparkSubmit() throws IOException {
File validSparkHome =
Files.createTempDirectory("gravitino-test-spark-home").toFile();
diff --git a/docs/table-maintenance-service/optimizer-cli-reference.md
b/docs/table-maintenance-service/optimizer-cli-reference.md
index f01ace00cc..82fdec1ecc 100644
--- a/docs/table-maintenance-service/optimizer-cli-reference.md
+++ b/docs/table-maintenance-service/optimizer-cli-reference.md
@@ -182,7 +182,10 @@ Notes:
- For `stats` or `all`, `--updater-options` must include `gravitino_uri` and
`metalake`.
- If `--updater-options` includes external JDBC metrics settings
(`gravitino.optimizer.jdbcMetrics.*`), ensure the JDBC driver JAR is
available to Spark
- runtime classpath (for example via `spark.jars` in `--spark-conf`).
+ runtime classpath of the driver: with the k8s job executor, for example via
`spark.jars` in
+ `--spark-conf`, which replaces a `spark.jars` default of the k8s job
executor, so list the
+ Iceberg runtime there too if that default provides it; with the local job
executor, by installing
+ it into `SPARK_HOME/jars`, as the driver applies `--spark-conf` only after
it starts.
- `--spark-conf` and `--updater-options` are flat JSON maps.
## List Table Metrics
@@ -229,7 +232,7 @@ Examples:
SUMMARY: statistics totalRecords=3 tableRecords=2 partitionRecords=1
jobRecords=0
DRY-RUN: strategy=iceberg-data-compaction identifier=rest_catalog.db.t1
score=95 jobTemplate=builtin-iceberg-rewrite-data-files
jobOptions={catalog_name=rest_catalog, table_identifier=db.t1}
SUBMIT: strategy=iceberg-data-compaction identifier=rest_catalog.db.t1
score=95 jobTemplate=builtin-iceberg-rewrite-data-files
jobOptions={catalog_name=rest_catalog, table_identifier=db.t1}
jobId=1f54c6d3-4e27-4cc8-bdfa-b05ecf59a4c2
-DRY-RUN: identifier=rest_catalog.db.t1
jobTemplate=builtin-iceberg-update-stats jobConfig={catalog_name=rest_catalog,
table_identifier=db.t1, update_mode=all,
updater_options={"gravitino_uri":"http://localhost:8090","metalake":"test"},
spark_conf={"spark.master":"local[2]","spark.hadoop.fs.defaultFS":"file:///"}}
+DRY-RUN: identifier=rest_catalog.db.t1
jobTemplate=builtin-iceberg-update-stats jobConfig={catalog_name=rest_catalog,
table_identifier=db.t1, update_mode=all,
updater_options={"gravitino_uri":"http://localhost:8090","metalake":"test"},
spark_conf={"spark.hadoop.fs.defaultFS":"file:///"}}
SUMMARY: submit-update-stats-job total=1 submitted=1 dryRun=false
MetricsResult{scopeType=TABLE, identifier=rest_catalog.db.t1,
partitionPath=<table-or-job-scope>, metrics={row_count=[{timestamp=1735689600,
value=100}]}}
EvaluationResult{scopeType=TABLE, identifier=rest_catalog.db.t1,
partitionPath=<table-or-job-scope>, evaluation=true,
evaluatorName=gravitino-metrics-evaluator, actionTimeSeconds=1735689600,
rangeSeconds=86400,
beforeMetrics={row_count=[MetricSample{timestampSeconds=1735686000,
value=120}]},
afterMetrics={row_count=[MetricSample{timestampSeconds=1735689600, value=100}]}}
@@ -254,7 +257,12 @@ These templates set Iceberg Spark session and catalog
classes, but they do not l
runtime in `jars`. `gravitino-jobs` also excludes that runtime from its shaded
JAR, so the version
that runs with your Spark cluster is yours to supply. Provide a matching
`iceberg-spark-runtime-<sparkMajor>_<scala>` JAR on the Spark classpath used
by the job executor —
-commonly through `spark.jars` in `spark_conf`, or by installing it into
`SPARK_HOME`. Align the
+commonly by installing it into `SPARK_HOME` or the Spark image, or, with the
k8s job executor,
+through `spark.jars` in `spark_conf`, which the k8s job executor sets when it
submits the job. It
+replaces a `spark.jars` default of the k8s job executor rather than adding to
it, so list every jar
+the job needs there.
+With the local job executor, `spark.jars` in `spark_conf` doesn't add it, as
the driver applies
+`spark_conf` only after spark-submit has built its classpath. Align the
artifact with the Spark, Scala, and Iceberg versions you actually run. A
reference coordinate used
in Gravitino's own jobs tests is
`org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.11.0`.
Without that runtime, built-in Iceberg jobs fail after Spark starts instead of
continuing without
@@ -265,18 +273,25 @@ left out, and a submission that misses a required key is
rejected with an error
missing keys. An empty string is a value, not an omission: it overrides the
default and means
"not set" to these jobs.
-| Job template | Required keys
| Optional keys and defaults
|
-|--------------------------------------|-----------------------------------------------------------------------------------------|--------------------------------------------------------------------------------------------------------------------|
-| Every Iceberg job | `catalog_name`, `table_identifier`,
`catalog_type`, `catalog_uri`, `warehouse_location` | `spark_master`
(`local[*]`), `spark_executor_instances` (`1`), `spark_executor_cores` (`1`),
`spark_executor_memory` (`1g`), `spark_driver_memory` (`1g`), `spark_conf`
(empty) |
-| `builtin-iceberg-update-stats` | —
| `update_mode` (`all`),
`updater_options` (empty)
|
-| `builtin-iceberg-rewrite-data-files` | `where_clause` (pass `""` to rewrite
the whole table) | `strategy` (`binpack`),
`sort_order` (empty), `options` (empty)
|
-| `builtin-iceberg-expire-snapshots` | —
| `older_than` (empty),
`retain_last` (empty), `stream_results` (`false`)
|
-| `builtin-iceberg-remove-orphan-files` | —
| `older_than` (empty),
`location` (empty), `dry_run` (`false`)
|
-| `builtin-iceberg-rewrite-manifests` | —
| `use_caching` (empty),
`spec_id` (empty)
|
+| Job template | Required keys
| Optional keys and defaults
|
+|---------------------------------------|-----------------------------------------------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| Every Iceberg job | `catalog_name`, `table_identifier`,
`catalog_type`, `catalog_uri`, `warehouse_location` | `spark_master` (empty),
`spark_executor_instances` (`1`), `spark_executor_cores` (`1`),
`spark_executor_memory` (`1g`), `spark_driver_memory` (`1g`), `spark_conf`
(empty) |
+| `builtin-iceberg-update-stats` | —
| `update_mode` (`all`),
`updater_options` (empty)
|
+| `builtin-iceberg-rewrite-data-files` | `where_clause` (pass `""` to rewrite
the whole table) | `strategy` (`binpack`),
`sort_order` (empty), `options` (empty)
|
+| `builtin-iceberg-expire-snapshots` | —
| `older_than` (empty),
`retain_last` (empty), `stream_results` (`false`)
|
+| `builtin-iceberg-remove-orphan-files` | —
| `older_than` (empty),
`location` (empty), `dry_run` (`false`)
|
+| `builtin-iceberg-rewrite-manifests` | —
| `use_caching` (empty),
`spec_id` (empty)
|
`where_clause` has no default on purpose: an empty where clause compacts every
data file in the
table, so it must be asked for explicitly.
+An empty `spark_master` leaves the master to the job executor: the local job
executor runs the job
+with the spark-submit default `local[*]`, and the k8s job executor rejects any
other master than a
+Kubernetes one. `spark.master` and `spark.submit.deployMode` in `spark_conf`
are ignored, as the job
+is submitted with them already set. So are the `spark.kubernetes.*`
configurations in `spark_conf`,
+such as the namespace, the image, the service account, Secrets, and volumes:
whoever runs a job sets
+`spark_conf`, so only the job executor configurations and job templates may
set them.
+
## Update Statistics
`builtin-iceberg-update-stats` reads a table and writes back the statistics
and metrics that policies evaluate. Compaction policies read
`custom-data-file-mse` and `custom-delete-file-number`, so nothing else will
fire until this job has run at least once.
diff --git a/docs/table-maintenance-service/optimizer-configuration.md
b/docs/table-maintenance-service/optimizer-configuration.md
index bd9bd65b46..a7618665b0 100644
--- a/docs/table-maintenance-service/optimizer-configuration.md
+++ b/docs/table-maintenance-service/optimizer-configuration.md
@@ -50,7 +50,7 @@ gravitino.optimizer.jobSubmitterConfig.catalog_type = rest
gravitino.optimizer.jobSubmitterConfig.catalog_uri =
http://localhost:9001/iceberg
# Leave empty for a local filesystem; set to your warehouse URI for cloud or
HDFS storage.
gravitino.optimizer.jobSubmitterConfig.warehouse_location =
-gravitino.optimizer.jobSubmitterConfig.spark_conf =
{"spark.master":"local[2]","spark.hadoop.fs.defaultFS":"file:///"}
+gravitino.optimizer.jobSubmitterConfig.spark_conf =
{"spark.hadoop.fs.defaultFS":"file:///"}
```
For manifest rewrite policies, the mapping
@@ -96,7 +96,7 @@ A direct job submission carries its own `jobConf`. This is
`builtin-iceberg-upda
"table_identifier": "db.t1",
"update_mode": "all",
"updater_options":
"{\"gravitino_uri\":\"http://localhost:8090\",\"metalake\":\"test\",\"statistics_updater\":\"gravitino-statistics-updater\",\"metrics_updater\":\"gravitino-metrics-updater\",\"auth_type\":\"basic\",\"username\":\"admin\",\"password\":\"YourSecureGravitinoPassword\"}";,
- "spark_conf":
"{\"spark.master\":\"local[2]\",\"spark.hadoop.fs.defaultFS\":\"file:///\"}",
+ "spark_conf": "{\"spark.hadoop.fs.defaultFS\":\"file:///\"}",
"spark_master": "local[2]",
"spark_executor_instances": "1",
"spark_executor_cores": "1",
@@ -117,8 +117,12 @@ that lists the missing keys. See
and the defaults of each template.
Built-in Iceberg templates also need an Iceberg Spark runtime on the Spark
classpath. They do not
-ship that JAR or fill template `jars`, so include it yourself — for example
-`"spark.jars":"/path/to/iceberg-spark-runtime-....jar"` inside `spark_conf`.
Match the artifact to
+ship that JAR or fill template `jars`, so include it yourself — for example by
installing it into
+`SPARK_HOME/jars`, or, with the k8s job executor, by
+`"spark.jars":"https://.../iceberg-spark-runtime-....jar"` inside
`spark_conf`, which replaces a
+`spark.jars` default of the k8s job executor, so list every jar the job needs
there. With the local job
+executor, `spark.jars` inside `spark_conf` doesn't add it, as the driver
applies `spark_conf` only
+after it starts. Match the artifact to
your Spark, Scala, and Iceberg versions. Details are under
[Built-in Job Templates](./optimizer-cli-reference.md#built-in-job-templates).
diff --git a/docs/table-maintenance-service/optimizer-troubleshooting.md
b/docs/table-maintenance-service/optimizer-troubleshooting.md
index 98e738021b..744ecf867d 100644
--- a/docs/table-maintenance-service/optimizer-troubleshooting.md
+++ b/docs/table-maintenance-service/optimizer-troubleshooting.md
@@ -64,14 +64,17 @@ These are the hard ones, because success and "the policy
decided not to act" loo
spark.hadoop.fs.defaultFS=file:///
```
-**`submit-update-stats-job` fails with JDBC metrics errors** — when
`--updater-options` includes `gravitino.optimizer.jdbcMetrics.*`, the JDBC
driver has to be on the Spark runtime classpath. `ClassNotFoundException` and
`No suitable driver` both mean the same thing:
+**`submit-update-stats-job` fails with JDBC metrics errors** — when
`--updater-options` includes `gravitino.optimizer.jdbcMetrics.*`, the JDBC
driver has to be on the Spark runtime classpath. `ClassNotFoundException` and
`No suitable driver` both mean the same thing. With the k8s job executor, add
it in `--spark-conf`; with the local job executor, install it into
`SPARK_HOME/jars`, as the driver applies `--spark-conf` only after it starts:
```json
{
- "spark.jars": "/path/to/postgresql-42.7.4.jar"
+ "spark.jars":
"https://repo1.maven.org/maven2/org/postgresql/postgresql/42.7.4/postgresql-42.7.4.jar";
}
```
+`spark.jars` in `spark_conf` replaces a `spark.jars` default of the k8s job
executor, so if that
+default provides the Iceberg runtime, list it in `spark_conf` as well.
+
**`The provided credentials did not support` or Iceberg `Not authorized`** — a
secured Gravitino endpoint needs `auth_type` / `username` / `password` (or
OAuth client-credentials fields) in `updater_options` for
`builtin-iceberg-update-stats`. A secured Iceberg REST catalog needs
`rest.auth.*` in `spark-conf` for any built-in job that reads the table. See
[Configuration](./optimizer-configuration.md).
**Built-in Iceberg jobs fail with `Missing Iceberg Spark session extensions`**
—
@@ -79,14 +82,18 @@ Spark only warns when `IcebergSparkSessionExtensions` is
missing, so built-in jo
classpath after `SparkSession` starts and exit with a non-zero status when the
Iceberg Spark
runtime is absent. The templates configure Iceberg classes but leave `jars`
empty, and
`gravitino-jobs` does not bundle `iceberg-spark-runtime`. A stock Spark
install is not enough.
-Put a version-matched Iceberg Spark runtime on the job classpath, for example:
+Put a version-matched Iceberg Spark runtime on the job classpath: install it
into `SPARK_HOME/jars`
+or the Spark image, or, with the k8s job executor, set it in `spark_conf`:
```json
{
- "spark.jars": "/path/to/iceberg-spark-runtime-3.5_2.12-1.11.0.jar"
+ "spark.jars":
"https://repo1.maven.org/maven2/org/apache/iceberg/iceberg-spark-runtime-3.5_2.12/1.11.0/iceberg-spark-runtime-3.5_2.12-1.11.0.jar";
}
```
+With the local job executor, `spark.jars` in `spark_conf` doesn't add it: the
driver applies
+`spark_conf` only after spark-submit has built its classpath.
+
Use the Spark, Scala, and Iceberg versions that match your cluster. See
[Built-in Job Templates](./optimizer-cli-reference.md#built-in-job-templates).
diff --git a/docs/table-maintenance-service/optimizer.md
b/docs/table-maintenance-service/optimizer.md
index 5048ccb693..eb4d651877 100644
--- a/docs/table-maintenance-service/optimizer.md
+++ b/docs/table-maintenance-service/optimizer.md
@@ -89,8 +89,9 @@ Each step ends with a check. If a check fails, stop there,
since every step depe
`IcebergSparkSessionExtensions` and `SparkCatalog`, but `gravitino-jobs`
does not ship the
Iceberg Spark runtime and the templates leave `jars` empty so your Spark and
Iceberg versions
stay under your control. A stock Spark distribution is not enough. Put a
matching
- `iceberg-spark-runtime-*` JAR on the job classpath — for example with
`spark.jars` in
- `spark_conf`, or by installing it into your Spark environment. Pick the
artifact that matches
+ `iceberg-spark-runtime-*` JAR on the job classpath — for example by
installing it into
+ `SPARK_HOME/jars`. `spark.jars` in `spark_conf` doesn't add it with the
local job executor, as
+ the driver applies `spark_conf` only after it starts. Pick the artifact that
matches
your Spark, Scala, and Iceberg versions. The jobs module is built and tested
against Spark 3.5.x
and Iceberg 1.11.0 (for example
`org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.11.0`). Without it, the
job fails after
@@ -222,7 +223,7 @@ update_stats_job_id=$(curl -sS -X POST -H "Accept:
application/vnd.gravitino.v1+
"table_identifier": "db.t1",
"update_mode": "all",
"updater_options":
"{\"gravitino_uri\":\"http://localhost:8090\",\"metalake\":\"test\",\"statistics_updater\":\"gravitino-statistics-updater\",\"metrics_updater\":\"gravitino-metrics-updater\"}";,
- "spark_conf":
"{\"spark.master\":\"local[2]\",\"spark.hadoop.fs.defaultFS\":\"file:///\"}",
+ "spark_conf": "{\"spark.hadoop.fs.defaultFS\":\"file:///\"}",
"spark_master": "local[2]",
"spark_executor_instances": "1",
"spark_executor_cores": "1",
@@ -270,7 +271,7 @@ gravitino.optimizer.jobSubmitterConfig.catalog_type = rest
gravitino.optimizer.jobSubmitterConfig.catalog_uri =
http://localhost:9001/iceberg
# Leave empty for a local filesystem; set to your warehouse URI for cloud or
HDFS storage.
gravitino.optimizer.jobSubmitterConfig.warehouse_location =
-gravitino.optimizer.jobSubmitterConfig.spark_conf =
{"spark.master":"local[2]","spark.hadoop.fs.defaultFS":"file:///"}
+gravitino.optimizer.jobSubmitterConfig.spark_conf =
{"spark.hadoop.fs.defaultFS":"file:///"}
EOF_CONF
```
diff --git
a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergExpireSnapshotsJob.java
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergExpireSnapshotsJob.java
index 1d80a76029..922f008c39 100644
---
a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergExpireSnapshotsJob.java
+++
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergExpireSnapshotsJob.java
@@ -39,7 +39,7 @@ public class IcebergExpireSnapshotsJob implements BuiltInJob {
private static final String NAME =
JobTemplateProvider.BUILTIN_NAME_PREFIX + "iceberg-expire-snapshots";
- private static final String VERSION = "v2";
+ private static final String VERSION = "v3";
@Override
public SparkJobTemplate jobTemplate() {
diff --git
a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergJobUtils.java
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergJobUtils.java
index 0340be3c33..0ac789d174 100644
---
a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergJobUtils.java
+++
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergJobUtils.java
@@ -18,13 +18,21 @@
*/
package org.apache.gravitino.maintenance.jobs.iceberg;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;
import java.util.regex.Pattern;
import javax.annotation.Nullable;
+import org.apache.commons.lang3.StringUtils;
import
org.apache.gravitino.maintenance.optimizer.common.util.IcebergSparkConfigUtils;
+import org.apache.spark.SparkConf;
import org.apache.spark.sql.SparkSession;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Shared utility methods for Iceberg maintenance jobs.
@@ -34,6 +42,8 @@ import org.apache.spark.sql.SparkSession;
*/
public final class IcebergJobUtils {
+ private static final Logger LOG =
LoggerFactory.getLogger(IcebergJobUtils.class);
+
private static final String ICEBERG_SPARK_CATALOG =
"org.apache.iceberg.spark.SparkCatalog";
/**
* Matches a job template placeholder that no job configuration value
replaced, e.g. {@code
@@ -41,6 +51,26 @@ public final class IcebergJobUtils {
*/
private static final Pattern UNRESOLVED_PLACEHOLDER_PATTERN =
Pattern.compile("^\\{\\{[^{}]*}}$");
+ /**
+ * The Spark configurations that spark-submit decides. Set in the custom
Spark configurations of a
+ * job, which the driver applies, they would override the submitted value,
e.g. run a job in local
+ * mode inside the driver pod of a cluster.
+ *
+ * <p>Keep this and {@link #SPARK_KUBERNETES_PREFIX} aligned with the k8s
job executor's {@code
+ * SparkApplicationUtils}, which ignores the same configurations of
spark_conf when it submits a
+ * built-in job. The jobs jar runs in the driver without the executor
plugin, so the rules can't
+ * be shared as code.
+ */
+ private static final Set<String> SUBMIT_DECIDED_CONFIGS =
+ ImmutableSet.of("spark.master", "spark.submit.deployMode");
+
+ /**
+ * The prefix of the Kubernetes configurations, which a job run may not set
in its custom Spark
+ * configurations: the driver creates the executor pods from them, with
their image, service
+ * account, Secrets, and volumes. Only the job executor configurations and
job templates set them.
+ */
+ private static final String SPARK_KUBERNETES_PREFIX = "spark.kubernetes.";
+
private IcebergJobUtils() {}
/**
@@ -173,7 +203,61 @@ public final class IcebergJobUtils {
* @throws IllegalArgumentException if JSON parsing fails
*/
public static Map<String, String> parseCustomSparkConfigs(String
sparkConfJson) {
- return IcebergSparkConfigUtils.parseFlatJsonMap(sparkConfJson,
"spark-conf");
+ return filterCustomSparkConfigs(
+ IcebergSparkConfigUtils.parseFlatJsonMap(sparkConfJson, "spark-conf"));
+ }
+
+ /**
+ * Filter the custom Spark configurations of a job, which the driver applies
when it creates the
+ * SparkSession, against the configurations spark-submit has already set for
the driver.
+ *
+ * <p>{@code spark.master} and {@code spark.submit.deployMode} are dropped
if spark-submit has set
+ * them, as the job would otherwise run with another master than it was
submitted to. The {@code
+ * spark.kubernetes.*} configurations are dropped, as whoever runs a job
sets them here, while the
+ * driver creates the executor pods from them. A blank value leaves a
configuration unset.
+ *
+ * @param configs the custom Spark configurations
+ * @return the custom Spark configurations to apply
+ */
+ public static Map<String, String> filterCustomSparkConfigs(Map<String,
String> configs) {
+ // new SparkConf() reads the spark.* system properties on purpose:
spark-submit sets the
+ // configurations it submitted the job with as system properties before it
calls main().
+ return filterCustomSparkConfigs(configs, new SparkConf());
+ }
+
+ @VisibleForTesting
+ static Map<String, String> filterCustomSparkConfigs(
+ Map<String, String> configs, SparkConf submittedConf) {
+ Map<String, String> filtered = new LinkedHashMap<>();
+ configs.forEach(
+ (key, value) -> {
+ // A blank value leaves a configuration unset, as the k8s job
executor does.
+ if (StringUtils.isBlank(value)) {
+ return;
+ }
+ if (key.startsWith(SPARK_KUBERNETES_PREFIX)) {
+ // The value isn't logged, as it may name a Secret.
+ LOG.warn(
+ "Ignored {} of the custom Spark configurations, as only the
job executor"
+ + " configurations and job templates may set {}*",
+ key,
+ SPARK_KUBERNETES_PREFIX);
+ return;
+ }
+ if (SUBMIT_DECIDED_CONFIGS.contains(key) &&
submittedConf.contains(key)) {
+ LOG.warn(
+ "Ignored {}={} of the custom Spark configurations, as the job
is submitted with"
+ + " {}={}. Set it when submitting the job instead, for a
built-in job by the"
+ + " spark_master parameter.",
+ key,
+ value,
+ key,
+ submittedConf.get(key));
+ return;
+ }
+ filtered.put(key, value);
+ });
+ return ImmutableMap.copyOf(filtered);
}
/**
@@ -181,8 +265,7 @@ public final class IcebergJobUtils {
*
* <p>Built-in templates configure {@code IcebergSparkSessionExtensions} and
{@code SparkCatalog},
* but Spark only warns when those classes are missing and continues without
Iceberg support. Call
- * this after {@code SparkSession} creation (so {@code spark.jars} from
{@code spark_conf} is
- * visible) and fail the job when the runtime is absent.
+ * this after {@code SparkSession} creation and fail the job when the
runtime is absent.
*
* @throws IllegalStateException when required Iceberg Spark classes cannot
be loaded
*/
@@ -233,8 +316,11 @@ public final class IcebergJobUtils {
throw new IllegalStateException(
String.format(
"Missing %s (%s). Built-in Iceberg jobs need
iceberg-spark-runtime on the Spark "
- + "classpath (for example via spark.jars in spark_conf, or
installed into the "
- + "Spark environment). A stock Spark distribution does not
include it. Match "
+ + "classpath, for example installed into the Spark
environment or the Spark "
+ + "image. With the k8s job executor, spark.jars in
spark_conf or in "
+ + "gravitino.jobExecutor.k8s.spark.conf.* adds it too. With
the local job "
+ + "executor, spark.jars in spark_conf doesn't, as the driver
applies spark_conf "
+ + "only after it starts. A stock Spark distribution does not
include it. Match "
+ "the artifact to your Spark, Scala, and Iceberg versions.",
description, className),
first);
diff --git
a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRemoveOrphanFilesJob.java
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRemoveOrphanFilesJob.java
index 85c84075e8..3c2e7f4e5d 100644
---
a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRemoveOrphanFilesJob.java
+++
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRemoveOrphanFilesJob.java
@@ -52,7 +52,7 @@ public class IcebergRemoveOrphanFilesJob implements
BuiltInJob {
private static final Logger LOG =
LoggerFactory.getLogger(IcebergRemoveOrphanFilesJob.class);
private static final String NAME =
JobTemplateProvider.BUILTIN_NAME_PREFIX + "iceberg-remove-orphan-files";
- private static final String VERSION = "v2";
+ private static final String VERSION = "v3";
@Override
public SparkJobTemplate jobTemplate() {
diff --git
a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteDataFilesJob.java
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteDataFilesJob.java
index 87ca6ab715..000b4da4e2 100644
---
a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteDataFilesJob.java
+++
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteDataFilesJob.java
@@ -39,7 +39,7 @@ public class IcebergRewriteDataFilesJob implements BuiltInJob
{
private static final String NAME =
JobTemplateProvider.BUILTIN_NAME_PREFIX + "iceberg-rewrite-data-files";
- private static final String VERSION = "v2";
+ private static final String VERSION = "v3";
// Valid strategy values for Iceberg rewrite_data_files procedure
private static final String STRATEGY_BINPACK = "binpack";
diff --git
a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java
index 4c4ec7dbd8..25d6f2f509 100644
---
a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java
+++
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergRewriteManifestsJob.java
@@ -41,7 +41,7 @@ public class IcebergRewriteManifestsJob implements BuiltInJob
{
private static final String NAME =
JobTemplateProvider.BUILTIN_NAME_PREFIX + "iceberg-rewrite-manifests";
- private static final String VERSION = "v2";
+ private static final String VERSION = "v3";
@Override
public SparkJobTemplate jobTemplate() {
diff --git
a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergUpdateManifestStatsJob.java
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergUpdateManifestStatsJob.java
index 2f398aa7d4..1b5d7553c2 100644
---
a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergUpdateManifestStatsJob.java
+++
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergUpdateManifestStatsJob.java
@@ -87,7 +87,7 @@ public class IcebergUpdateManifestStatsJob implements
BuiltInJob {
"--spark-conf",
"{{spark_conf:-}}"))
.withConfigs(IcebergSparkConfigUtils.buildTemplateSparkConfigs())
-
.withCustomFields(Collections.singletonMap(JobTemplateProvider.PROPERTY_VERSION_KEY,
"v1"))
+
.withCustomFields(Collections.singletonMap(JobTemplateProvider.PROPERTY_VERSION_KEY,
"v2"))
.build();
}
@@ -118,7 +118,7 @@ public class IcebergUpdateManifestStatsJob implements
BuiltInJob {
arguments.get("updater-options"), "updater-options");
Map<String, String> properties = buildOptimizerProperties(options);
Map<String, String> sparkConfigs =
- IcebergSparkConfigUtils.parseFlatJsonMap(arguments.get("spark-conf"),
"spark-conf");
+ IcebergJobUtils.parseCustomSparkConfigs(arguments.get("spark-conf"));
try (StatisticsUpdater updater =
ProviderUtils.createStatisticsUpdaterInstance(
options.getOrDefault("statistics_updater",
DEFAULT_STATISTICS_UPDATER).trim())) {
diff --git
a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergUpdateStatsAndMetricsJob.java
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergUpdateStatsAndMetricsJob.java
index abf482de3b..843b589553 100644
---
a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergUpdateStatsAndMetricsJob.java
+++
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergUpdateStatsAndMetricsJob.java
@@ -61,7 +61,7 @@ public class IcebergUpdateStatsAndMetricsJob implements
BuiltInJob {
private static final String NAME =
JobTemplateProvider.BUILTIN_NAME_PREFIX + "iceberg-update-stats";
- private static final String VERSION = "v2";
+ private static final String VERSION = "v3";
private static final String DEFAULT_STATISTICS_UPDATER =
"gravitino-statistics-updater";
private static final String DEFAULT_METRICS_UPDATER =
"gravitino-metrics-updater";
private static final long DEFAULT_TARGET_FILE_SIZE_BYTES = 128L * 1024 *
1024;
@@ -394,7 +394,7 @@ public class IcebergUpdateStatsAndMetricsJob implements
BuiltInJob {
@VisibleForTesting
static Map<String, String> parseCustomSparkConfigs(String sparkConfJson) {
- return parseJsonOptions(sparkConfJson, "spark-conf");
+ return
IcebergJobUtils.filterCustomSparkConfigs(parseJsonOptions(sparkConfJson,
"spark-conf"));
}
/**
diff --git
a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/spark/SparkPiJob.java
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/spark/SparkPiJob.java
index e97348bb5e..83cb951aca 100644
---
a/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/spark/SparkPiJob.java
+++
b/maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/spark/SparkPiJob.java
@@ -40,7 +40,7 @@ public class SparkPiJob implements BuiltInJob {
private static final String NAME = JobTemplateProvider.BUILTIN_NAME_PREFIX +
"sparkpi";
// Bump VERSION whenever SparkPi template behavior changes
(name/executable/class/args/configs).
- private static final String VERSION = "v2";
+ private static final String VERSION = "v3";
@Override
public SparkJobTemplate jobTemplate() {
@@ -94,7 +94,9 @@ public class SparkPiJob implements BuiltInJob {
private Map<String, String> buildSparkConfigs() {
Map<String, String> configs = new HashMap<>();
- configs.put("spark.master", "{{spark_master:-local[*]}}");
+ // An empty value leaves the Spark configuration unset, see
+ // IcebergSparkConfigUtils#buildTemplateSparkConfigs.
+ configs.put("spark.master", "{{spark_master:-}}");
configs.put("spark.executor.instances", "{{spark_executor_instances:-1}}");
configs.put("spark.executor.cores", "{{spark_executor_cores:-1}}");
configs.put("spark.executor.memory", "{{spark_executor_memory:-1g}}");
diff --git
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergExpireSnapshotsJob.java
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergExpireSnapshotsJob.java
index 1795704542..a0344d8301 100644
---
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergExpireSnapshotsJob.java
+++
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergExpireSnapshotsJob.java
@@ -133,7 +133,7 @@ public class TestIcebergExpireSnapshotsJob {
assertTrue(customFields.containsKey(JobTemplateProvider.PROPERTY_VERSION_KEY));
String version =
customFields.get(JobTemplateProvider.PROPERTY_VERSION_KEY);
- assertEquals("v2", version);
+ assertEquals("v3", version);
assertTrue(version.matches(JobTemplateProvider.VERSION_VALUE_PATTERN));
}
diff --git
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergJobUtils.java
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergJobUtils.java
index 89fb606e3d..867ec45591 100644
---
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergJobUtils.java
+++
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergJobUtils.java
@@ -24,11 +24,13 @@ import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import com.google.common.collect.ImmutableMap;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
+import org.apache.spark.SparkConf;
import org.junit.jupiter.api.Test;
public class TestIcebergJobUtils {
@@ -99,4 +101,45 @@ public class TestIcebergJobUtils {
() -> IcebergJobUtils.parseCustomSparkConfigs("{not_json}"));
assertTrue(ex.getMessage().contains("--spark-conf"));
}
+
+ @Test
+ public void testFilterCustomSparkConfigsDropsSubmittedMasterAndDeployMode() {
+ Map<String, String> configs =
+ ImmutableMap.of(
+ "spark.master", "local[*]",
+ "spark.submit.deployMode", "client",
+ "spark.jars", "/opt/iceberg-spark-runtime.jar",
+ "spark.sql.shuffle.partitions", "200");
+ SparkConf submitted =
+ new SparkConf(false)
+ .set("spark.master", "k8s://https://kubernetes.default.svc";)
+ .set("spark.submit.deployMode", "cluster");
+
+ assertEquals(
+ ImmutableMap.of(
+ "spark.jars", "/opt/iceberg-spark-runtime.jar",
"spark.sql.shuffle.partitions", "200"),
+ IcebergJobUtils.filterCustomSparkConfigs(configs, submitted));
+ }
+
+ @Test
+ public void testFilterCustomSparkConfigsKeepsMasterNotSubmitted() {
+ // A job run without spark-submit, e.g. in a test, has no master unless it
sets one.
+ Map<String, String> configs = ImmutableMap.of("spark.master", "local[1]");
+ assertEquals(configs, IcebergJobUtils.filterCustomSparkConfigs(configs,
new SparkConf(false)));
+ }
+
+ @Test
+ public void testFilterCustomSparkConfigsDropsKubernetesAndBlankConfigs() {
+ Map<String, String> configs =
+ ImmutableMap.of(
+ "spark.kubernetes.executor.secretKeyRef.TOKEN", "secret:token",
+ "spark.kubernetes.container.image", "other:1.0",
+ "spark.sql.shuffle.partitions", "",
+ // Blank like StringUtils.isBlank, as the k8s job executor checks
it: an em space.
+ "spark.driver.memory", "\u2003",
+ "spark.executor.memory", "2g");
+ assertEquals(
+ ImmutableMap.of("spark.executor.memory", "2g"),
+ IcebergJobUtils.filterCustomSparkConfigs(configs, new
SparkConf(false)));
+ }
}
diff --git
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteDataFilesJob.java
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteDataFilesJob.java
index e734991fb6..7aec76aad0 100644
---
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteDataFilesJob.java
+++
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteDataFilesJob.java
@@ -134,7 +134,7 @@ public class TestIcebergRewriteDataFilesJob {
assertTrue(customFields.containsKey(JobTemplateProvider.PROPERTY_VERSION_KEY));
String version =
customFields.get(JobTemplateProvider.PROPERTY_VERSION_KEY);
- assertEquals("v2", version);
+ assertEquals("v3", version);
assertTrue(version.matches(JobTemplateProvider.VERSION_VALUE_PATTERN));
}
diff --git
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJob.java
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJob.java
index 9d5f894acf..7f819d1d41 100644
---
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJob.java
+++
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergRewriteManifestsJob.java
@@ -197,7 +197,7 @@ public class TestIcebergRewriteManifestsJob {
assertTrue(customFields.containsKey(JobTemplateProvider.PROPERTY_VERSION_KEY));
String version =
customFields.get(JobTemplateProvider.PROPERTY_VERSION_KEY);
- assertEquals("v2", version);
+ assertEquals("v3", version);
assertTrue(version.matches(JobTemplateProvider.VERSION_VALUE_PATTERN));
}
diff --git
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergUpdateManifestStatsJob.java
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergUpdateManifestStatsJob.java
index 24b1efe30e..71d26bd8c6 100644
---
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergUpdateManifestStatsJob.java
+++
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergUpdateManifestStatsJob.java
@@ -64,7 +64,7 @@ public class TestIcebergUpdateManifestStatsJob {
assertTrue(template.arguments().contains("{{spark_conf:-}}"));
assertTrue(template.arguments().contains("{{coordination_directory:-}}"));
assertFalse(template.arguments().contains("--update-mode"));
- assertEquals("v1", template.customFields().get("version"));
+ assertEquals("v2", template.customFields().get("version"));
assertTrue(template.configs().containsKey("spark.sql.extensions"));
SparkJobTemplate existing = (SparkJobTemplate)
templates.get("builtin-iceberg-update-stats");
assertNotNull(existing);
diff --git
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergUpdateStatsJob.java
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergUpdateStatsJob.java
index d15ddb92ca..3e7d6a488d 100644
---
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergUpdateStatsJob.java
+++
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/iceberg/TestIcebergUpdateStatsJob.java
@@ -42,7 +42,7 @@ public class TestIcebergUpdateStatsJob {
assertNotNull(template);
assertEquals("builtin-iceberg-update-stats", template.name());
assertTrue(template.name().matches(JobTemplateProvider.BUILTIN_NAME_PATTERN));
- assertEquals("v2",
template.customFields().get(JobTemplateProvider.PROPERTY_VERSION_KEY));
+ assertEquals("v3",
template.customFields().get(JobTemplateProvider.PROPERTY_VERSION_KEY));
}
@Test
diff --git
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/spark/TestSparkPiJob.java
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/spark/TestSparkPiJob.java
index 768f3c881a..c0d468a673 100644
---
a/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/spark/TestSparkPiJob.java
+++
b/maintenance/jobs/src/test/java/org/apache/gravitino/maintenance/jobs/spark/TestSparkPiJob.java
@@ -93,7 +93,7 @@ public class TestSparkPiJob {
assertTrue(configs.containsKey("spark.driver.memory"));
// Verify placeholders
- assertEquals("{{spark_master:-local[*]}}", configs.get("spark.master"));
+ assertEquals("{{spark_master:-}}", configs.get("spark.master"));
assertEquals("{{spark_executor_instances:-1}}",
configs.get("spark.executor.instances"));
assertEquals("{{spark_executor_cores:-1}}",
configs.get("spark.executor.cores"));
assertEquals("{{spark_executor_memory:-1g}}",
configs.get("spark.executor.memory"));
@@ -110,7 +110,7 @@ public class TestSparkPiJob {
assertTrue(customFields.containsKey(JobTemplateProvider.PROPERTY_VERSION_KEY));
String version =
customFields.get(JobTemplateProvider.PROPERTY_VERSION_KEY);
- assertEquals("v2", version);
+ assertEquals("v3", version);
assertTrue(version.matches(JobTemplateProvider.VERSION_VALUE_PATTERN));
}
diff --git
a/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/common/util/IcebergSparkConfigUtils.java
b/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/common/util/IcebergSparkConfigUtils.java
index a1f92aae7e..b0e9c60983 100644
---
a/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/common/util/IcebergSparkConfigUtils.java
+++
b/maintenance/optimizer-api/src/main/java/org/apache/gravitino/maintenance/optimizer/common/util/IcebergSparkConfigUtils.java
@@ -50,7 +50,9 @@ public final class IcebergSparkConfigUtils {
/** Build default Spark template configs for Iceberg jobs. */
public static Map<String, String> buildTemplateSparkConfigs() {
Map<String, String> configs = new HashMap<>();
- configs.put("spark.master", "{{spark_master:-local[*]}}");
+ // An empty value leaves the Spark configuration unset: spark-submit then
runs the job with
+ // its default master, local[*], and the k8s job executor lets the
operator decide it.
+ configs.put("spark.master", "{{spark_master:-}}");
configs.put("spark.executor.instances", "{{spark_executor_instances:-1}}");
configs.put("spark.executor.cores", "{{spark_executor_cores:-1}}");
configs.put("spark.executor.memory", "{{spark_executor_memory:-1g}}");
diff --git
a/maintenance/optimizer-api/src/test/java/org/apache/gravitino/maintenance/optimizer/common/util/TestIcebergSparkConfigUtils.java
b/maintenance/optimizer-api/src/test/java/org/apache/gravitino/maintenance/optimizer/common/util/TestIcebergSparkConfigUtils.java
index 0b4da3ea25..c2dead444b 100644
---
a/maintenance/optimizer-api/src/test/java/org/apache/gravitino/maintenance/optimizer/common/util/TestIcebergSparkConfigUtils.java
+++
b/maintenance/optimizer-api/src/test/java/org/apache/gravitino/maintenance/optimizer/common/util/TestIcebergSparkConfigUtils.java
@@ -46,6 +46,13 @@ public class TestIcebergSparkConfigUtils {
IcebergSparkConfigUtils.ICEBERG_SPARK_EXTENSIONS,
configs.get("spark.sql.extensions"));
}
+ @Test
+ public void testBuildTemplateSparkConfigsLeavesMasterUnsetByDefault() {
+ Map<String, String> configs =
IcebergSparkConfigUtils.buildTemplateSparkConfigs();
+
+ assertEquals("{{spark_master:-}}", configs.get("spark.master"));
+ }
+
@Test
public void testParseFlatJsonMap() {
Map<String, String> parsed =
diff --git
a/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/command/SubmitUpdateStatsJobCommand.java
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/command/SubmitUpdateStatsJobCommand.java
index 1f08d8dfd6..10ef3402a4 100644
---
a/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/command/SubmitUpdateStatsJobCommand.java
+++
b/maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/command/SubmitUpdateStatsJobCommand.java
@@ -47,7 +47,6 @@ public class SubmitUpdateStatsJobCommand implements
OptimizerCommandExecutor {
private static final String DEFAULT_UPDATE_MODE = "all";
private static final String OPTION_UPDATER_OPTIONS = "updater-options";
private static final String OPTION_SPARK_CONF = "spark-conf";
- private static final String DEFAULT_SPARK_MASTER = "local[*]";
private static final String DEFAULT_SPARK_EXECUTOR_INSTANCES = "1";
private static final String DEFAULT_SPARK_EXECUTOR_CORES = "1";
private static final String DEFAULT_SPARK_EXECUTOR_MEMORY = "1g";
@@ -98,6 +97,17 @@ public class SubmitUpdateStatsJobCommand implements
OptimizerCommandExecutor {
validateUpdaterOptions(updateMode, updaterOptions);
validateSparkConfigs(tableTargets, sparkConfigs);
+ // The job ignores the master of spark_conf, as spark-submit decides it,
so say where it goes.
+ if (StringUtils.isNotBlank(sparkConfigs.get(SPARK_MASTER_KEY))) {
+ context
+ .output()
+ .printf(
+ "WARN: %s=%s in the Spark configurations is ignored by the job.
Set %sspark_master"
+ + " instead.%n",
+ SPARK_MASTER_KEY,
+ sparkConfigs.get(SPARK_MASTER_KEY),
+ OptimizerConfig.JOB_SUBMITTER_CONFIG_PREFIX);
+ }
if (context.dryRun()) {
for (TableTarget tableTarget : tableTargets) {
@@ -333,12 +343,10 @@ public class SubmitUpdateStatsJobCommand implements
OptimizerCommandExecutor {
Map<String, String> placeholders = new LinkedHashMap<>();
String catalogPrefix = SPARK_SQL_CATALOG_PREFIX + catalogName;
- placeholders.put(
- "spark_master",
- firstNonBlank(
- submitterConfigs.get("spark_master"),
- sparkConfigs.get(SPARK_MASTER_KEY),
- DEFAULT_SPARK_MASTER));
+ // An empty master leaves it to the job executor: spark-submit defaults to
local[*], and the
+ // k8s job executor lets the operator decide it. spark.master in
spark_conf isn't used, as the
+ // job ignores it there too.
+ placeholders.put("spark_master",
firstNonBlank(submitterConfigs.get("spark_master")));
placeholders.put(
"spark_executor_instances",
firstNonBlank(
diff --git
a/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/TestOptimizerCmd.java
b/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/TestOptimizerCmd.java
index 8ea3772db5..2a05a0b417 100644
---
a/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/TestOptimizerCmd.java
+++
b/maintenance/optimizer/src/test/java/org/apache/gravitino/maintenance/optimizer/TestOptimizerCmd.java
@@ -26,6 +26,7 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.List;
+import java.util.regex.Pattern;
import org.apache.gravitino.NameIdentifier;
import
org.apache.gravitino.maintenance.optimizer.monitor.evaluator.MetricsEvaluatorForTest;
import
org.apache.gravitino.maintenance.optimizer.monitor.job.TableJobRelationProviderForTest;
@@ -42,6 +43,9 @@ import org.junit.jupiter.api.Test;
class TestOptimizerCmd {
+ // A spark_master entry with an empty value in the printed job configuration.
+ private static final Pattern EMPTY_SPARK_MASTER =
Pattern.compile("\\bspark_master=\\s*[,}]");
+
@Test
void testGlobalHelpIncludesImplementedCommands() {
String[] output = runCommand("--help");
@@ -497,7 +501,8 @@ class TestOptimizerCmd {
Assertions.assertTrue(output[0].contains("jobTemplate=builtin-iceberg-update-stats"));
Assertions.assertTrue(output[0].contains("update_mode=stats"));
Assertions.assertTrue(output[0].contains("table_identifier=ab.t1"));
- Assertions.assertTrue(output[0].contains("spark_master=local[*]"));
+ // Without a configured master, the job executor decides it.
+ Assertions.assertTrue(EMPTY_SPARK_MASTER.matcher(output[0]).find(),
output[0]);
Assertions.assertTrue(output[0].contains("catalog_type=rest"));
Assertions.assertTrue(output[0].contains("catalog_uri=http://localhost:9001/iceberg";));
Assertions.assertTrue(output[0].contains("warehouse_location=/tmp/warehouse"));
@@ -554,13 +559,21 @@ class TestOptimizerCmd {
+ "\"spark.sql.catalog.rest.type\":\"rest\","
+
"\"spark.sql.catalog.rest.uri\":\"http://localhost:9001/iceberg\",";
+ "\"spark.sql.catalog.rest.warehouse\":\"/tmp/warehouse\","
- +
"\"spark.sql.extensions\":\"org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions\"}",
+ +
"\"spark.sql.extensions\":\"org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions\","
+ + "\"spark.master\":\"local[2]\"}",
"--dry-run",
"--conf-path",
confPath.toString());
Assertions.assertTrue(output[1].isEmpty(), "stderr=" + output[1] + ",
stdout=" + output[0]);
Assertions.assertTrue(output[0].contains("custom-metrics-updater"));
Assertions.assertTrue(output[0].contains("spark.sql.catalog.rest"));
+ // The master of spark_conf isn't passed as the spark_master parameter,
and the user is told.
+ Assertions.assertTrue(EMPTY_SPARK_MASTER.matcher(output[0]).find(),
output[0]);
+ Assertions.assertTrue(
+ output[0].contains(
+ "WARN: spark.master=local[2] in the Spark configurations is
ignored by the job."
+ + " Set gravitino.optimizer.jobSubmitterConfig.spark_master
instead."),
+ output[0]);
}
@Test
diff --git a/plugins/k8s-job-executor/build.gradle.kts
b/plugins/k8s-job-executor/build.gradle.kts
index ec06621218..495bf59b93 100644
--- a/plugins/k8s-job-executor/build.gradle.kts
+++ b/plugins/k8s-job-executor/build.gradle.kts
@@ -48,6 +48,9 @@ dependencies {
exclude(group = "io.fabric8", module = "kubernetes-httpclient-vertx")
}
implementation(libs.fabric8.kubernetes.httpclient.jdk)
+ // Parses the spark_conf of built-in jobs. Resolved to the version fabric8
needs, and relocated
+ // with it.
+ implementation(libs.jackson.databind)
testImplementation(project(":api"))
testImplementation(project(":common"))
diff --git
a/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/spark/SparkApplicationUtils.java
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/spark/SparkApplicationUtils.java
index 8ce2a8a6d2..53b170480c 100644
---
a/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/spark/SparkApplicationUtils.java
+++
b/plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/spark/SparkApplicationUtils.java
@@ -18,8 +18,12 @@
*/
package org.apache.gravitino.job.k8s.spark;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
import io.fabric8.kubernetes.api.model.GenericKubernetesResource;
import io.fabric8.kubernetes.api.model.GenericKubernetesResourceBuilder;
import io.fabric8.kubernetes.client.dsl.base.ResourceDefinitionContext;
@@ -28,6 +32,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
+import java.util.Set;
import java.util.TreeMap;
import java.util.regex.Pattern;
import org.apache.commons.lang3.StringUtils;
@@ -80,6 +85,7 @@ public final class SparkApplicationUtils {
private static final String SPARK_MASTER = "spark.master";
private static final String SPARK_DEPLOY_MODE = "spark.submit.deployMode";
private static final String CLUSTER_DEPLOY_MODE = "cluster";
+ private static final String K8S_MASTER_PREFIX = "k8s://";
private static final String SPARK_IMAGE = "spark.kubernetes.container.image";
private static final String SPARK_DRIVER_SERVICE_ACCOUNT =
"spark.kubernetes.authenticate.driver.serviceAccountName";
@@ -89,6 +95,22 @@ public final class SparkApplicationUtils {
private static final String SPARK_DRIVER_ENV_PREFIX =
"spark.kubernetes.driverEnv.";
private static final String SPARK_EXECUTOR_ENV_PREFIX = "spark.executorEnv.";
+ // The argument of a built-in job that carries the Spark configurations of
the job run.
+ private static final String BUILTIN_SPARK_CONF_ARGUMENT = "--spark-conf";
+
+ // The configurations that spark-submit decides, which a built-in job
ignores in its spark_conf.
+ // Keep this and SPARK_KUBERNETES_PREFIX aligned with IcebergJobUtils of the
built-in jobs, whose
+ // driver ignores the same configurations of spark_conf; the jobs jar can't
depend on this plugin.
+ private static final Set<String> SUBMIT_DECIDED_CONFIGS =
+ ImmutableSet.of(SPARK_MASTER, SPARK_DEPLOY_MODE);
+
+ // The prefix of the Kubernetes configurations, such as the namespace, the
image, the service
+ // account, Secrets, and volumes. Whoever can run a built-in job sets its
spark_conf, so only the
+ // job executor configurations and job templates may set them.
+ private static final String SPARK_KUBERNETES_PREFIX = "spark.kubernetes.";
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
// The environment variable names that Spark passes to the executors: Spark
3.5 drops any other
// spark.executorEnv.* key with only a warning, while the driver would still
get it. Kubernetes
// without the relaxed environment variable validation accepts the same
names, except '.' and
@@ -111,7 +133,13 @@ public final class SparkApplicationUtils {
* through the Spark configurations {@code spark.kubernetes.driverEnv.*} and
{@code
* spark.executorEnv.*}, which means their values are visible in the
SparkApplication and in its
* pods. They override the default Spark configurations of the job executor,
and the Spark
- * configurations of the job template override them.
+ * configurations of the job override them, except those with a blank value,
which are left unset.
+ * The Spark configurations of a built-in job include those of its {@code
spark_conf} parameter,
+ * so that the configurations that spark-submit reads, such as {@code
spark.jars}, take effect.
+ *
+ * <p>The operator decides the master and the deploy mode, so a job whose
{@code spark.master}
+ * isn't a Kubernetes master, or whose {@code spark.submit.deployMode} isn't
{@code cluster}, is
+ * rejected.
*
* @param context the context of the job run
* @param template the runtime job template, whose resources are the URIs
given in the template
@@ -136,7 +164,8 @@ public final class SparkApplicationUtils {
sparkConf.put(SPARK_DRIVER_ENV_PREFIX + name, value);
sparkConf.put(SPARK_EXECUTOR_ENV_PREFIX + name, value);
});
- sparkConf.putAll(template.configs());
+ Map<String, String> jobSparkConf = getJobSparkConf(context, template);
+ sparkConf.putAll(jobSparkConf);
// The operator only supports cluster mode, so a job that asks for client
mode is rejected
// rather than silently run in cluster mode.
String deployMode =
StringUtils.trimToNull(sparkConf.get(SPARK_DEPLOY_MODE));
@@ -147,6 +176,21 @@ public final class SparkApplicationUtils {
+ " the k8s job executor only runs Spark jobs in %s mode",
SPARK_DEPLOY_MODE, deployMode, template.name(),
CLUSTER_DEPLOY_MODE));
}
+ // Likewise, a job that asks for another master, such as local[*], is
rejected rather than
+ // silently run on Kubernetes.
+ String master = StringUtils.trimToNull(sparkConf.get(SPARK_MASTER));
+ if (master != null && !master.startsWith(K8S_MASTER_PREFIX)) {
+ String hint =
+ template.name().startsWith(JobTemplateProvider.BUILTIN_NAME_PREFIX)
+ ? " For a built-in job template, leave the spark_master
parameter empty."
+ : "";
+ throw new IllegalArgumentException(
+ String.format(
+ "%s is set to %s by job template %s or by the k8s job executor
configurations, but"
+ + " the k8s job executor only runs Spark jobs on Kubernetes.
Leave %s unset in"
+ + " the job template and in the k8s job executor
configurations.%s",
+ SPARK_MASTER, master, template.name(), SPARK_MASTER, hint));
+ }
// The operator decides the master and the deploy mode.
sparkConf.remove(SPARK_MASTER);
sparkConf.remove(SPARK_DEPLOY_MODE);
@@ -154,7 +198,7 @@ public final class SparkApplicationUtils {
putIfBlank(sparkConf, SPARK_DRIVER_SERVICE_ACCOUNT,
configs.sparkServiceAccount());
// Only a job chooses its namespace this way. A default set in the Spark
configurations of
// the executor would silently shadow the namespace of the executor
configurations.
- String namespace =
StringUtils.trimToNull(template.configs().get(SPARK_NAMESPACE));
+ String namespace =
StringUtils.trimToNull(jobSparkConf.get(SPARK_NAMESPACE));
if (namespace == null) {
namespace = configs.namespace();
} else if (!K8sJobResourceUtils.isValidNamespace(namespace)) {
@@ -216,6 +260,112 @@ public final class SparkApplicationUtils {
return app;
}
+ /**
+ * Returns the Spark configurations that a job sets itself, which override
the default Spark
+ * configurations of the job executor.
+ *
+ * <p>They are the configurations of the job template, and for a built-in
job also those of its
+ * {@code spark_conf} parameter, which override them. The driver of a
built-in job applies its
+ * {@code spark_conf} only once it starts, too late for the configurations
that spark-submit
+ * reads, such as {@code spark.jars}, so they are set in the
SparkApplication as well. {@code
+ * spark.master} and {@code spark.submit.deployMode} there are ignored, as
the driver also ignores
+ * them, and so are the {@code spark.kubernetes.*} configurations, which
only the job executor
+ * configurations and job templates may set: whoever can run a built-in job
sets its {@code
+ * spark_conf}. A blank value leaves a configuration unset, as it is what an
optional template
+ * parameter becomes, so it doesn't override the default of the job executor.
+ */
+ private static Map<String, String> getJobSparkConf(
+ JobContext context, SparkJobTemplate template) {
+ Map<String, String> jobSparkConf = new TreeMap<>();
+ template
+ .configs()
+ .forEach(
+ (key, value) -> {
+ if (StringUtils.isNotBlank(value)) {
+ checkExecutorEnvironmentName(template, key);
+ jobSparkConf.put(key, value);
+ }
+ });
+ if (!template.name().startsWith(JobTemplateProvider.BUILTIN_NAME_PREFIX)) {
+ return jobSparkConf;
+ }
+
+ getBuiltinSparkConf(template)
+ .forEach(
+ (key, value) -> {
+ if (StringUtils.isBlank(value)) {
+ return;
+ }
+ if (SUBMIT_DECIDED_CONFIGS.contains(key)) {
+ LOG.warn(
+ "Ignored {}={} in spark_conf of job {} of built-in job
template {}, as the"
+ + " Spark operator decides it",
+ key,
+ value,
+ context.jobId(),
+ template.name());
+ } else if (key.startsWith(SPARK_KUBERNETES_PREFIX)) {
+ // The value isn't logged, as it may name a Secret.
+ LOG.warn(
+ "Ignored {} in spark_conf of job {} of built-in job
template {}, as only the"
+ + " k8s job executor configurations and job templates
may set {}*",
+ key,
+ context.jobId(),
+ template.name(),
+ SPARK_KUBERNETES_PREFIX);
+ } else {
+ checkExecutorEnvironmentName(template, key);
+ jobSparkConf.put(key, value);
+ }
+ });
+ return jobSparkConf;
+ }
+
+ private static Map<String, String> getBuiltinSparkConf(SparkJobTemplate
template) {
+ List<String> arguments = template.arguments();
+ // A built-in template has the argument once. A value of another argument
that equals it would
+ // be followed by a value that isn't JSON, which rejects the job rather
than skipping a check.
+ int index = arguments.indexOf(BUILTIN_SPARK_CONF_ARGUMENT);
+ if (index < 0 || index + 1 >= arguments.size()) {
+ return ImmutableMap.of();
+ }
+ String json = arguments.get(index + 1);
+ if (StringUtils.isBlank(json)) {
+ return ImmutableMap.of();
+ }
+
+ Map<String, Object> parsed;
+ try {
+ parsed = MAPPER.readValue(json, new TypeReference<Map<String, Object>>()
{});
+ } catch (JsonProcessingException e) {
+ throw new IllegalArgumentException(
+ String.format(
+ "spark_conf of built-in job template %s isn't a JSON map: %s",
+ template.name(), e.getOriginalMessage()),
+ e);
+ }
+ // The built-in jobs reject a JSON null as well.
+ if (parsed == null) {
+ throw new IllegalArgumentException(
+ String.format(
+ "spark_conf of built-in job template %s isn't a JSON map: null",
template.name()));
+ }
+ // Parsed the way the built-in jobs parse it: a flat map whose null values
are empty.
+ Map<String, String> sparkConf = new LinkedHashMap<>();
+ parsed.forEach(
+ (key, value) -> {
+ if (value instanceof Map || value instanceof List) {
+ throw new IllegalArgumentException(
+ String.format(
+ "spark_conf of built-in job template %s must be a flat
JSON map, but %s has"
+ + " a non-scalar value",
+ template.name(), key));
+ }
+ sparkConf.put(key, value == null ? "" : value.toString());
+ });
+ return ImmutableMap.copyOf(sparkConf);
+ }
+
private static String getExecutableUri(
JobContext context, SparkJobTemplate template, K8sJobExecutorConfigs
configs) {
// The executable of a built-in job template is the path of the jobs jar
on the Gravitino
@@ -238,6 +388,14 @@ public final class SparkApplicationUtils {
return template.executable();
}
+ // Checks the environment variable name of a spark.executorEnv.*
configuration like those of the
+ // environments of the template, as an invalid name would only fail the
executor pods.
+ private static void checkExecutorEnvironmentName(SparkJobTemplate template,
String key) {
+ if (key.startsWith(SPARK_EXECUTOR_ENV_PREFIX)) {
+ checkEnvironmentName(template,
key.substring(SPARK_EXECUTOR_ENV_PREFIX.length()));
+ }
+ }
+
private static void checkEnvironmentName(SparkJobTemplate template, String
name) {
// Checked here, as an invalid name would otherwise only miss the
executors, or fail the job
// when the operator creates the driver pod.
diff --git
a/plugins/k8s-job-executor/src/shadedJarTest/java/org/apache/gravitino/job/k8s/TestK8sJobExecutorShadedJar.java
b/plugins/k8s-job-executor/src/shadedJarTest/java/org/apache/gravitino/job/k8s/TestK8sJobExecutorShadedJar.java
index 023f930d91..facdd1fe7e 100644
---
a/plugins/k8s-job-executor/src/shadedJarTest/java/org/apache/gravitino/job/k8s/TestK8sJobExecutorShadedJar.java
+++
b/plugins/k8s-job-executor/src/shadedJarTest/java/org/apache/gravitino/job/k8s/TestK8sJobExecutorShadedJar.java
@@ -169,7 +169,8 @@ public class TestK8sJobExecutorShadedJar {
.withName("builtin-sparkpi")
.withExecutable("/gravitino/auxlib/gravitino-jobs.jar")
.withClassName("org.apache.gravitino.maintenance.jobs.spark.SparkPiJob")
- .withConfigs(ImmutableMap.of("spark.master", "local[*]"))
+ // What the built-in template's {{spark_master:-}} resolves to
by default.
+ .withConfigs(ImmutableMap.of("spark.master", ""))
.build();
// The factory finds the built-in k8s job executor by its class name.
diff --git
a/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/spark/TestSparkApplicationUtils.java
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/spark/TestSparkApplicationUtils.java
index 39da670461..ee4a2b9d3d 100644
---
a/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/spark/TestSparkApplicationUtils.java
+++
b/plugins/k8s-job-executor/src/test/java/org/apache/gravitino/job/k8s/spark/TestSparkApplicationUtils.java
@@ -23,6 +23,7 @@ import com.google.common.collect.ImmutableMap;
import io.fabric8.kubernetes.api.model.GenericKubernetesResource;
import java.io.File;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import org.apache.gravitino.connector.job.JobContext;
import org.apache.gravitino.job.SparkJobTemplate;
@@ -67,6 +68,85 @@ public class TestSparkApplicationUtils {
() -> SparkApplicationUtils.buildSparkApplication(CONTEXT, noMode,
clientMode));
}
+ @Test
+ public void testRejectNonKubernetesMaster() {
+ for (String master : new String[] {"local[*]", " local ", "yarn",
"spark://host:7077"}) {
+ SparkJobTemplate template =
+ newTemplate("s3a://bucket/app.jar", ImmutableMap.of("spark.master",
master));
+ IllegalArgumentException e =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> SparkApplicationUtils.buildSparkApplication(CONTEXT,
template, configs));
+ Assertions.assertTrue(e.getMessage().contains(master.trim()),
e.getMessage());
+ // The spark_master parameter only exists in the built-in job templates.
+ Assertions.assertFalse(e.getMessage().contains("spark_master"),
e.getMessage());
+ }
+
+ SparkJobTemplate builtin =
+ newTemplateBuilder()
+ .withName("builtin-sparkpi")
+ .withConfigs(ImmutableMap.of("spark.master", "local[*]"))
+ .build();
+ IllegalArgumentException builtinError =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> SparkApplicationUtils.buildSparkApplication(CONTEXT,
builtin, configs));
+ Assertions.assertTrue(
+ builtinError.getMessage().contains("spark_master"),
builtinError.getMessage());
+
+ // Nor set by the executor configurations.
+ K8sJobExecutorConfigs localMaster =
+ newConfigs(ImmutableMap.of("spark.conf.spark.master", "local[2]"));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ SparkApplicationUtils.buildSparkApplication(
+ CONTEXT, newTemplate("s3a://bucket/app.jar",
ImmutableMap.of()), localMaster));
+ }
+
+ @Test
+ public void testIgnoreBlankMaster() {
+ // A blank master, as the built-in templates set by default, is left to
the operator.
+ Map<String, Object> spec =
+ getSpec(
+ newTemplate(
+ "s3a://bucket/app.jar",
+ ImmutableMap.of("spark.master", " ",
"spark.submit.deployMode", "")),
+ configs);
+ @SuppressWarnings("unchecked")
+ Map<String, String> sparkConf = (Map<String, String>)
spec.get("sparkConf");
+ Assertions.assertFalse(sparkConf.containsKey("spark.master"));
+ Assertions.assertFalse(sparkConf.containsKey("spark.submit.deployMode"));
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testBlankConfigsKeepExecutorDefaults() {
+ K8sJobExecutorConfigs withDefaults =
+ newConfigs(
+ ImmutableMap.of(
+ "spark.conf.spark.jars",
"https://repo/iceberg-spark-runtime.jar";,
+ "spark.conf.spark.jars.packages", "org.example:lib:1.0"));
+ SparkJobTemplate template =
+ newTemplate(
+ "s3a://bucket/app.jar",
+ ImmutableMap.of(
+ "spark.jars", "", "spark.jars.packages", " ",
"spark.executor.memory", ""));
+
+ Map<String, String> sparkConf =
+ (Map<String, String>) getSpec(template, withDefaults).get("sparkConf");
+ Assertions.assertEquals("https://repo/iceberg-spark-runtime.jar";,
sparkConf.get("spark.jars"));
+ Assertions.assertEquals("org.example:lib:1.0",
sparkConf.get("spark.jars.packages"));
+ Assertions.assertFalse(sparkConf.containsKey("spark.executor.memory"));
+
+ // A value set by the job overrides the executor default.
+ SparkJobTemplate withJars =
+ newTemplate(
+ "s3a://bucket/app.jar", ImmutableMap.of("spark.jars",
"https://repo/other.jar";));
+ sparkConf = (Map<String, String>) getSpec(withJars,
withDefaults).get("sparkConf");
+ Assertions.assertEquals("https://repo/other.jar";,
sparkConf.get("spark.jars"));
+ }
+
@Test
@SuppressWarnings("unchecked")
public void testBuildSparkApplication() {
@@ -81,7 +161,7 @@ public class TestSparkApplicationUtils {
.withArchives(ImmutableList.of("hdfs://nn/env.tgz#env"))
.withConfigs(
ImmutableMap.of(
- "spark.master", "local[*]",
+ "spark.master", "k8s://https://kubernetes.default.svc";,
"spark.submit.deployMode", "cluster",
"spark.executor.instances", "2",
"spark.jars", "https://repo/other.jar";))
@@ -492,6 +572,203 @@ public class TestSparkApplicationUtils {
"https://repo/gravitino-jobs.jar";, getSpec(template,
configs).get("jars"));
}
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testSparkConfOfBuiltinJob() {
+ String json =
+ "{\"spark.jars\":\"https://repo/iceberg-spark-runtime.jar\",";
+ + "\"spark.jars.packages\":\"org.example:lib:1.0\","
+ + "\"spark.executor.memory\":\"4g\","
+ + "\"spark.sql.shuffle.partitions\":200,"
+ + "\"spark.master\":\"local[*]\","
+ + "\"spark.submit.deployMode\":\"client\"}";
+ List<String> arguments = ImmutableList.of("--catalog", "demo",
"--spark-conf", json);
+ SparkJobTemplate template =
+ newTemplateBuilder()
+ .withName("builtin-iceberg-rewrite-data-files")
+ .withArguments(arguments)
+ .withConfigs(
+ ImmutableMap.of(
+ "spark.master", "",
+ "spark.executor.memory", "1g",
+ "spark.driver.memory", "1g"))
+ .build();
+
+ Map<String, Object> spec = getSpec(template, configs);
+ Map<String, String> sparkConf = (Map<String, String>)
spec.get("sparkConf");
+ Assertions.assertEquals("https://repo/iceberg-spark-runtime.jar";,
sparkConf.get("spark.jars"));
+ Assertions.assertEquals("org.example:lib:1.0",
sparkConf.get("spark.jars.packages"));
+ // spark_conf overrides the configurations of the template.
+ Assertions.assertEquals("4g", sparkConf.get("spark.executor.memory"));
+ Assertions.assertEquals("1g", sparkConf.get("spark.driver.memory"));
+ Assertions.assertEquals("200",
sparkConf.get("spark.sql.shuffle.partitions"));
+ // The master and the deploy mode of spark_conf are ignored rather than
rejected.
+ Assertions.assertFalse(sparkConf.containsKey("spark.master"));
+ Assertions.assertFalse(sparkConf.containsKey("spark.submit.deployMode"));
+ // The driver still gets spark_conf.
+ Assertions.assertEquals(arguments, spec.get("driverArgs"));
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testSparkConfOfBuiltinJobKeepsExecutorDefaults() {
+ K8sJobExecutorConfigs withDefaults =
+ newConfigs(
+ ImmutableMap.of(
+ K8sJobExecutorConfigs.NAMESPACE,
+ "jobs",
+ "spark.conf.spark.jars",
+ "https://repo/iceberg-spark-runtime.jar";));
+ for (String json : new String[] {"", " ", "{}", "{\"spark.jars\":\" \"}"})
{
+ SparkJobTemplate template =
+ newTemplateBuilder()
+ .withName("builtin-iceberg-expire-snapshots")
+ .withArguments(ImmutableList.of("--spark-conf", json))
+ .build();
+ Map<String, String> sparkConf =
+ (Map<String, String>) getSpec(template,
withDefaults).get("sparkConf");
+ Assertions.assertEquals(
+ "https://repo/iceberg-spark-runtime.jar";,
sparkConf.get("spark.jars"), json);
+ Assertions.assertEquals("jobs",
sparkConf.get("spark.kubernetes.namespace"), json);
+ }
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testSparkConfOfBuiltinJobReplacesExecutorDefaults() {
+ K8sJobExecutorConfigs withDefaults =
+ newConfigs(
+ ImmutableMap.of(
+ K8sJobExecutorConfigs.NAMESPACE,
+ "jobs",
+ "spark.conf.spark.jars",
+ "https://repo/iceberg-spark-runtime.jar";));
+ SparkJobTemplate template =
+ newTemplateBuilder()
+ .withName("builtin-iceberg-update-stats")
+ .withArguments(
+ ImmutableList.of(
+ "--spark-conf",
"{\"spark.jars\":\"https://repo/postgresql.jar\"}";))
+ .build();
+
+ // Like any Spark configuration of the job, spark.jars replaces the
executor default.
+ Map<String, String> sparkConf =
+ (Map<String, String>) getSpec(template, withDefaults).get("sparkConf");
+ Assertions.assertEquals("https://repo/postgresql.jar";,
sparkConf.get("spark.jars"));
+ }
+
+ @Test
+ public void testRejectInvalidEnvironmentNameInSparkConfOfBuiltinJob() {
+ SparkJobTemplate template =
+ newTemplateBuilder()
+ .withName("builtin-iceberg-update-stats")
+ .withArguments(
+ ImmutableList.of("--spark-conf", "{\"spark.executorEnv.BAD
NAME\":\"v\"}"))
+ .build();
+ IllegalArgumentException e =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> SparkApplicationUtils.buildSparkApplication(CONTEXT,
template, configs));
+ Assertions.assertTrue(e.getMessage().contains("BAD NAME"), e.getMessage());
+
+ // Nor in the configs of a template.
+ SparkJobTemplate inConfigs =
+ newTemplate("s3a://bucket/app.jar",
ImmutableMap.of("spark.executorEnv.BAD NAME", "v"));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> SparkApplicationUtils.buildSparkApplication(CONTEXT, inConfigs,
configs));
+
+ SparkJobTemplate valid =
+ newTemplateBuilder()
+ .withName("builtin-iceberg-update-stats")
+ .withArguments(ImmutableList.of("--spark-conf",
"{\"spark.executorEnv.MY_VAR\":\"v\"}"))
+ .build();
+ @SuppressWarnings("unchecked")
+ Map<String, String> sparkConf = (Map<String, String>) getSpec(valid,
configs).get("sparkConf");
+ Assertions.assertEquals("v", sparkConf.get("spark.executorEnv.MY_VAR"));
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testSparkConfOfBuiltinJobIgnoresKubernetesConfigs() {
+ String json =
+ "{\"spark.kubernetes.namespace\":\"kube-system\","
+ +
"\"spark.kubernetes.authenticate.driver.serviceAccountName\":\"admin\","
+ + "\"spark.kubernetes.container.image\":\"evil:1.0\","
+ +
"\"spark.kubernetes.driver.secretKeyRef.TOKEN\":\"secret:token\","
+ + "\"spark.executor.memory\":\"2g\"}";
+ SparkJobTemplate template =
+ newTemplateBuilder()
+ .withName("builtin-iceberg-expire-snapshots")
+ .withArguments(ImmutableList.of("--spark-conf", json))
+ .build();
+
+ GenericKubernetesResource app =
+ SparkApplicationUtils.buildSparkApplication(CONTEXT, template,
configs);
+
+ Assertions.assertEquals("default", app.getMetadata().getNamespace());
+ Map<String, String> sparkConf =
+ (Map<String, String>)
+ ((Map<String, Object>)
app.getAdditionalProperties().get("spec")).get("sparkConf");
+ Assertions.assertEquals("default",
sparkConf.get("spark.kubernetes.namespace"));
+ Assertions.assertEquals(
+ "spark",
sparkConf.get("spark.kubernetes.authenticate.driver.serviceAccountName"));
+ Assertions.assertEquals(
+ "gravitino-spark:3.5.9",
sparkConf.get("spark.kubernetes.container.image"));
+
Assertions.assertFalse(sparkConf.containsKey("spark.kubernetes.driver.secretKeyRef.TOKEN"));
+ Assertions.assertEquals("2g", sparkConf.get("spark.executor.memory"));
+ }
+
+ @Test
+ public void testRejectInvalidSparkConfOfBuiltinJob() {
+ for (String json :
+ new String[] {
+ "{bad}",
+ "null",
+ "\"a\"",
+ "[\"a\"]",
+ "{\"spark.jars\":[\"a.jar\"]}",
+ "{\"k\":{\"x\":\"y\"}}"
+ }) {
+ SparkJobTemplate template =
+ newTemplateBuilder()
+ .withName("builtin-iceberg-expire-snapshots")
+ .withArguments(ImmutableList.of("--spark-conf", json))
+ .build();
+ IllegalArgumentException e =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> SparkApplicationUtils.buildSparkApplication(CONTEXT,
template, configs),
+ json);
+ Assertions.assertTrue(e.getMessage().contains("spark_conf"),
e.getMessage());
+ }
+
+ // A jar on the Gravitino server can't be read by the pods.
+ SparkJobTemplate serverJar =
+ newTemplateBuilder()
+ .withName("builtin-iceberg-expire-snapshots")
+ .withArguments(
+ ImmutableList.of("--spark-conf",
"{\"spark.jars\":\"/opt/iceberg.jar\"}"))
+ .build();
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> SparkApplicationUtils.buildSparkApplication(CONTEXT, serverJar,
configs));
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testSparkConfArgumentOfOtherJobsIsNotParsed() {
+ SparkJobTemplate template =
+ newTemplateBuilder()
+ .withName("user-job")
+ .withArguments(
+ ImmutableList.of("--spark-conf",
"{\"spark.jars\":\"https://repo/a.jar\"}";))
+ .build();
+ Map<String, String> sparkConf =
+ (Map<String, String>) getSpec(template, configs).get("sparkConf");
+ Assertions.assertFalse(sparkConf.containsKey("spark.jars"));
+ }
+
@Test
public void testBuiltinJobsJarOnlyReplacesExecutableOfBuiltinJobs() {
// Not a built-in job: "builtin-" must be the prefix of the name.
