You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2025/03/16 22:09:47 UTC
(spark-kubernetes-operator) branch main updated: [SPARK-51507] Support creating config map that can be mounted by Spark pods for apps
This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/spark-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new aa76c01 [SPARK-51507] Support creating config map that can be mounted by Spark pods for apps
aa76c01 is described below
commit aa76c01d397652a833cf2b42745458be96fb7f18
Author: Zhou JIANG <ji...@umich.edu>
AuthorDate: Sun Mar 16 15:09:39 2025 -0700
[SPARK-51507] Support creating config map that can be mounted by Spark pods for apps
### What changes were proposed in this pull request?
This PR introduces a new field `configMapSpecs`, which enables user to create configmap(s) which can be later mounted on driver and/or executors with values specified in SparkApplications. It also adds corresponding docs and unit tests.
### Why are the changes needed?
There's a common user demand for the feature to create lightweight values / files override using configmap. This new feature saves the trouble for user to create configmap(s) beforehand - the config map would share the lifecycle of the corresponding Spark app.
User may mount the ConfigMap as needed in driver and executor pod templates - one example is provided in the doc. In future we may further enhance this flow by mounting the created config map automatically to Spark pods. This can be achieved by supporting volume type of ConfigMap in Spark Kubernetes, or by adding mutating webhook & alter pod templates after creation. For now, this PR opens up the possibility of creating config map and mounting them with one single spec.
### Does this PR introduce _any_ user-facing change?
No - not yet released
### How was this patch tested?
Pass the CIs
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #166 from jiangzho/cm.
Authored-by: Zhou JIANG <ji...@umich.edu>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
docs/spark_custom_resources.md | 56 +++++++++++++++++
.../spark/k8s/operator/spec/ApplicationSpec.java | 1 +
.../{ApplicationSpec.java => ConfigMapSpec.java} | 29 ++-------
.../spark/k8s/operator/SparkAppResourceSpec.java | 6 +-
.../k8s/operator/SparkAppSubmissionWorker.java | 10 +++-
.../k8s/operator/utils/ConfigMapSpecUtils.java | 51 ++++++++++++++++
.../k8s/operator/SparkAppResourceSpecTest.java | 3 +-
.../k8s/operator/utils/ConfigMapSpecUtilsTest.java | 70 ++++++++++++++++++++++
8 files changed, 199 insertions(+), 27 deletions(-)
diff --git a/docs/spark_custom_resources.md b/docs/spark_custom_resources.md
index 3a42c51..6a8751d 100644
--- a/docs/spark_custom_resources.md
+++ b/docs/spark_custom_resources.md
@@ -138,6 +138,62 @@ default rule backed by the associated Service. It's recommended to always provid
to make sure it's compatible with your
[IngressController](https://kubernetes.io/docs/concepts/services-networking/ingress-controllers/).
+## Create and Mount ConfigMap
+
+It is possible to ask operator to create configmap so they can be used by driver and/or executor
+pods on the fly. `configMapSpecs` allows you to specify the desired metadata and data as string
+literals for the configmap(s) to be created.
+
+```yaml
+spec:
+ configMapSpecs:
+ - name: "example-config-map"
+ data:
+ foo: "bar"
+```
+
+Like other app-specific resources, the created configmap has owner reference to Spark driver and
+therefore shares the same lifecycle and garbage collection mechanism with the associated app.
+
+This feature can be used to create lightweight override config files for given Spark app. For
+example, below snippet would create and mount a configmap with metrics property file, then use it
+in SparkConf:
+
+```yaml
+spec:
+ sparkConf:
+ spark.metrics.conf: "/etc/metrics/metrics.properties"
+ driverSpec:
+ podTemplateSpec:
+ spec:
+ containers:
+ - volumeMounts:
+ - name: "config-override"
+ mountPath: "/etc/metrics"
+ readOnly: true
+ volumes:
+ - name: config-override
+ configMap:
+ name: metrics-configmap
+ executorSpec:
+ podTemplateSpec:
+ spec:
+ containers:
+ - volumeMounts:
+ - name: "config-override"
+ mountPath: "/etc/metrics"
+ readOnly: true
+ volumes:
+ - name: config-override
+ configMap:
+ name: metrics-configmap
+ configMapSpecs:
+ - name: "metrics-configmap"
+ data:
+ metrics.properties: "*.sink.jmx.class=org.apache.spark.metrics.sink.JmxSink\n"
+
+```
+
## Understanding Failure Types
In addition to the general `Failed` state (that driver pod fails or driver container exits
diff --git a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/ApplicationSpec.java b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/ApplicationSpec.java
index 2a830f6..9d7722f 100644
--- a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/ApplicationSpec.java
+++ b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/ApplicationSpec.java
@@ -55,4 +55,5 @@ public class ApplicationSpec extends BaseSpec {
protected BaseApplicationTemplateSpec driverSpec;
protected BaseApplicationTemplateSpec executorSpec;
protected List<DriverServiceIngressSpec> driverServiceIngressList;
+ protected List<ConfigMapSpec> configMapSpecs;
}
diff --git a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/ApplicationSpec.java b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/ConfigMapSpec.java
similarity index 55%
copy from spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/ApplicationSpec.java
copy to spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/ConfigMapSpec.java
index 2a830f6..8cce3fd 100644
--- a/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/ApplicationSpec.java
+++ b/spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/ConfigMapSpec.java
@@ -19,40 +19,23 @@
package org.apache.spark.k8s.operator.spec;
-import java.util.ArrayList;
-import java.util.List;
+import java.util.Map;
-import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonInclude;
import io.fabric8.generator.annotation.Required;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
-import lombok.EqualsAndHashCode;
import lombok.NoArgsConstructor;
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
-@EqualsAndHashCode(callSuper = true)
@JsonInclude(JsonInclude.Include.NON_NULL)
-@JsonIgnoreProperties(ignoreUnknown = true)
-public class ApplicationSpec extends BaseSpec {
- protected String mainClass;
- @Required protected RuntimeVersions runtimeVersions;
- protected String jars;
- protected String pyFiles;
- protected String sparkRFiles;
- protected String files;
- @Builder.Default protected DeploymentMode deploymentMode = DeploymentMode.ClusterMode;
- protected String proxyUser;
- @Builder.Default protected List<String> driverArgs = new ArrayList<>();
-
- @Builder.Default
- protected ApplicationTolerations applicationTolerations = new ApplicationTolerations();
-
- protected BaseApplicationTemplateSpec driverSpec;
- protected BaseApplicationTemplateSpec executorSpec;
- protected List<DriverServiceIngressSpec> driverServiceIngressList;
+public class ConfigMapSpec {
+ @Required protected String name;
+ protected Map<String, String> labels;
+ protected Map<String, String> annotations;
+ protected Map<String, String> data;
}
diff --git a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppResourceSpec.java b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppResourceSpec.java
index 43d931a..8fe2bba 100644
--- a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppResourceSpec.java
+++ b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppResourceSpec.java
@@ -43,7 +43,9 @@ import org.apache.spark.deploy.k8s.Constants;
import org.apache.spark.deploy.k8s.KubernetesDriverSpec;
import org.apache.spark.deploy.k8s.SparkPod;
import org.apache.spark.deploy.k8s.submit.KubernetesClientUtils;
+import org.apache.spark.k8s.operator.spec.ConfigMapSpec;
import org.apache.spark.k8s.operator.spec.DriverServiceIngressSpec;
+import org.apache.spark.k8s.operator.utils.ConfigMapSpecUtils;
import org.apache.spark.k8s.operator.utils.DriverServiceIngressUtils;
/**
@@ -66,7 +68,8 @@ public class SparkAppResourceSpec {
public SparkAppResourceSpec(
SparkAppDriverConf kubernetesDriverConf,
KubernetesDriverSpec kubernetesDriverSpec,
- List<DriverServiceIngressSpec> driverServiceIngressList) {
+ List<DriverServiceIngressSpec> driverServiceIngressList,
+ List<ConfigMapSpec> configMapSpecs) {
this.kubernetesDriverConf = kubernetesDriverConf;
String namespace = kubernetesDriverConf.sparkConf().get(Config.KUBERNETES_NAMESPACE().key());
Map<String, String> confFilesMap =
@@ -93,6 +96,7 @@ public class SparkAppResourceSpec {
this.driverResources.add(
KubernetesClientUtils.buildConfigMap(
kubernetesDriverConf.configMapNameDriver(), confFilesMap, new HashMap<>()));
+ this.driverPreResources.addAll(ConfigMapSpecUtils.buildConfigMaps(configMapSpecs));
this.driverResources.addAll(configureDriverServerIngress(sparkPod, driverServiceIngressList));
this.driverPreResources.forEach(r -> setNamespaceIfMissing(r, namespace));
this.driverResources.forEach(r -> setNamespaceIfMissing(r, namespace));
diff --git a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppSubmissionWorker.java b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppSubmissionWorker.java
index 7cf78d1..d204c5d 100644
--- a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppSubmissionWorker.java
+++ b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppSubmissionWorker.java
@@ -38,6 +38,7 @@ import org.apache.spark.deploy.k8s.submit.MainAppResource;
import org.apache.spark.deploy.k8s.submit.PythonMainAppResource;
import org.apache.spark.deploy.k8s.submit.RMainAppResource;
import org.apache.spark.k8s.operator.spec.ApplicationSpec;
+import org.apache.spark.k8s.operator.spec.ConfigMapSpec;
import org.apache.spark.k8s.operator.spec.DriverServiceIngressSpec;
import org.apache.spark.k8s.operator.utils.ModelUtils;
@@ -84,7 +85,11 @@ public class SparkAppSubmissionWorker {
public SparkAppResourceSpec getResourceSpec(
SparkApplication app, KubernetesClient client, Map<String, String> confOverrides) {
SparkAppDriverConf appDriverConf = buildDriverConf(app, confOverrides);
- return buildResourceSpec(appDriverConf, app.getSpec().getDriverServiceIngressList(), client);
+ return buildResourceSpec(
+ appDriverConf,
+ app.getSpec().getDriverServiceIngressList(),
+ app.getSpec().getConfigMapSpecs(),
+ client);
}
protected SparkAppDriverConf buildDriverConf(
@@ -131,12 +136,13 @@ public class SparkAppSubmissionWorker {
protected SparkAppResourceSpec buildResourceSpec(
SparkAppDriverConf kubernetesDriverConf,
List<DriverServiceIngressSpec> driverServiceIngressList,
+ List<ConfigMapSpec> configMapSpecs,
KubernetesClient client) {
KubernetesDriverBuilder builder = new KubernetesDriverBuilder();
KubernetesDriverSpec kubernetesDriverSpec =
builder.buildFromFeatures(kubernetesDriverConf, client);
return new SparkAppResourceSpec(
- kubernetesDriverConf, kubernetesDriverSpec, driverServiceIngressList);
+ kubernetesDriverConf, kubernetesDriverSpec, driverServiceIngressList, configMapSpecs);
}
/**
diff --git a/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/utils/ConfigMapSpecUtils.java b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/utils/ConfigMapSpecUtils.java
new file mode 100644
index 0000000..fed1d1b
--- /dev/null
+++ b/spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/utils/ConfigMapSpecUtils.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.k8s.operator.utils;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.api.model.ConfigMapBuilder;
+
+import org.apache.spark.k8s.operator.spec.ConfigMapSpec;
+
+public final class ConfigMapSpecUtils {
+ private ConfigMapSpecUtils() {}
+
+ public static List<ConfigMap> buildConfigMaps(List<ConfigMapSpec> configMapMountList) {
+ if (configMapMountList == null || configMapMountList.isEmpty()) {
+ return Collections.emptyList();
+ }
+ return configMapMountList.stream()
+ .map(
+ c ->
+ new ConfigMapBuilder()
+ .editOrNewMetadata()
+ .withName(c.getName())
+ .withLabels(c.getLabels())
+ .withAnnotations(c.getAnnotations())
+ .endMetadata()
+ .addToData(c.getData())
+ .build())
+ .collect(Collectors.toList());
+ }
+}
diff --git a/spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/SparkAppResourceSpecTest.java b/spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/SparkAppResourceSpecTest.java
index 42e2106..e6a50a0 100644
--- a/spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/SparkAppResourceSpecTest.java
+++ b/spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/SparkAppResourceSpecTest.java
@@ -70,7 +70,8 @@ class SparkAppResourceSpecTest {
when(mockSpec.systemProperties()).thenReturn(new HashMap<>());
SparkAppResourceSpec appResourceSpec =
- new SparkAppResourceSpec(mockConf, mockSpec, Collections.emptyList());
+ new SparkAppResourceSpec(
+ mockConf, mockSpec, Collections.emptyList(), Collections.emptyList());
Assertions.assertEquals(2, appResourceSpec.getDriverResources().size());
Assertions.assertEquals(1, appResourceSpec.getDriverPreResources().size());
diff --git a/spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/utils/ConfigMapSpecUtilsTest.java b/spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/utils/ConfigMapSpecUtilsTest.java
new file mode 100644
index 0000000..deb1aa7
--- /dev/null
+++ b/spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/utils/ConfigMapSpecUtilsTest.java
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.k8s.operator.utils;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import org.junit.jupiter.api.Test;
+
+import org.apache.spark.k8s.operator.spec.ConfigMapSpec;
+
+class ConfigMapSpecUtilsTest {
+
+ @Test
+ void buildConfigMaps() {
+
+ assertEquals(Collections.emptyList(), ConfigMapSpecUtils.buildConfigMaps(null));
+ assertEquals(
+ Collections.emptyList(), ConfigMapSpecUtils.buildConfigMaps(Collections.emptyList()));
+
+ Map<String, String> labels = Map.of("foo-label-key", "foo-label-value");
+ Map<String, String> annotations = Map.of("foo-annotation-key", "foo-annotation-value");
+ Map<String, String> configMapData = new HashMap<>();
+ // literal value
+ configMapData.put("foo", "bar");
+ // escaped property file
+ configMapData.put(
+ "conf.properties", "spark.foo.keyOne=\"valueOne\"\nspark.foo.keyTwo=\"valueTwo\"");
+ // escaped yaml file
+ configMapData.put("conf.yaml", "conf:\n foo: bar\n nestedField:\n keyName: value");
+ ConfigMapSpec mount =
+ ConfigMapSpec.builder()
+ .name("test-config-map")
+ .labels(labels)
+ .annotations(annotations)
+ .data(configMapData)
+ .build();
+ List<ConfigMap> configMaps =
+ ConfigMapSpecUtils.buildConfigMaps(Collections.singletonList(mount));
+ assertEquals(1, configMaps.size());
+ ConfigMap configMap = configMaps.get(0);
+ assertEquals("test-config-map", configMap.getMetadata().getName());
+ assertEquals("foo-label-value", configMap.getMetadata().getLabels().get("foo-label-key"));
+ assertEquals(
+ "foo-annotation-value", configMap.getMetadata().getAnnotations().get("foo-annotation-key"));
+ assertEquals(configMapData, configMap.getData());
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org