jerryshao commented on code in PR #13982:
URL: https://github.com/apache/gravitino/pull/13982#discussion_r4332699517
##########
plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/spark/SparkApplicationUtils.java:
##########
@@ -216,6 +260,110 @@ public static GenericKubernetesResource
buildSparkApplication(
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)) {
+ 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 {
+ jobSparkConf.put(key, value);
Review Comment:
Kept the override on purpose: a non-blank Spark configuration of a job
replaces the executor default, the same rule as for the template's configs
(this was decided when we settled the precedence). In 2cd6eb975 the docs now
say it in optimizer-cli-reference.md (both places) and
optimizer-troubleshooting.md: spark.jars in spark_conf replaces the executor's
spark.jars default, so list the Iceberg runtime there too. Also added
testSparkConfOfBuiltinJobReplacesExecutorDefaults to pin the behavior.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
maintenance/optimizer/src/main/java/org/apache/gravitino/maintenance/optimizer/command/SubmitUpdateStatsJobCommand.java:
##########
@@ -333,12 +331,10 @@ private static Map<String, String>
resolveTemplateSparkPlaceholders(
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")));
Review Comment:
submit-update-stats-job now prints `WARN: spark.master=<value> in the Spark
configurations is ignored by the job. Set
gravitino.optimizer.jobSubmitterConfig.spark_master instead.` when spark_conf
sets a master. I chose a warning over a hard rejection so existing configs keep
submitting. TestOptimizerCmd asserts the message.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
plugins/k8s-job-executor/src/main/java/org/apache/gravitino/job/k8s/spark/SparkApplicationUtils.java:
##########
@@ -136,7 +164,8 @@ public static GenericKubernetesResource
buildSparkApplication(
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);
Review Comment:
spark.executorEnv.* keys from spark_conf now go through
checkEnvironmentName, the same as the template's environments, so an invalid
name rejects the run at submit time. Added
testRejectInvalidEnvironmentNameInSparkConfOfBuiltinJob (invalid name rejected,
valid name kept).
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
maintenance/jobs/src/main/java/org/apache/gravitino/maintenance/jobs/iceberg/IcebergJobUtils.java:
##########
@@ -173,16 +202,67 @@ public static Map<String, String> parseArguments(
* @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) {
+ return filterCustomSparkConfigs(configs, new SparkConf());
Review Comment:
Added a comment that the no-arg new SparkConf() reads the spark.* system
properties on purpose, because spark-submit sets the submitted configurations
as system properties before main(). The tests keep using the overload that
takes an explicit SparkConf.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
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;
Review Comment:
Declared implementation(libs.jackson.databind) with a comment.
dependencyInsight shows runtimeClasspath still resolves 2.18.3 (the 2.15.2
declaration is upgraded to fabric8's version), so the bundled and relocated
Jackson is unchanged. shadedJarTest passes.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
