persistant volume mount

[devops-eda]

hello hussein,
i am facing an issue with volume mounting . am unable to mount the volume can you please suggest me the correct way to mount the volume . so that I can pick a file from that mounting path while am submitting a python file.
below are the file currently am using .

import sys
from spark_on_k8s.client import SparkOnK8S, ExecutorInstances, PodResources

# Initialize the SparkOnK8S client
spark_client = SparkOnK8S()

# Configure Spark to access MinIO
spark_conf = {
    "spark.kubernetes.container.image.pullPolicy": "Always",
    "spark.kubernetes.volumes.persistentVolumeClaim.sparky-data.options.claimName": "cafebot-pvc",
    "spark.kubernetes.volumes.persistentVolumeClaim.sparky-data.mount.path": "/opt/cafebot/spark-host",
    "spark.kubernetes.volumes.persistentVolumeClaim.sparky-data.mount.subPath": "spark-host"
}

# Submit the Spark application
spark_client.submit_app(
    image="eda2019/spark-submition:poc-1.7",
    app_path="local:///opt/cafebot/spark-host/process_data.py",
    app_arguments=[],
    app_name="pyspark-job",
    namespace="cafebot",
    service_account="spark",
    app_waiter="log",
    ui_reverse_proxy=True,
    driver_resources=PodResources(cpu=1, memory=512, memory_overhead=128),
    executor_resources=PodResources(cpu=1, memory=512, memory_overhead=128),
    executor_instances=ExecutorInstances(initial=2),
    driver_volume_mounts="/opt/cafebot/spark-host",
    executor_volume_mounts="/opt/cafebot/spark-host",
    spark_conf=spark_conf  # Pass the Spark configuration
)

[hussein-awala]

You don't need to add the configuration explicitly, you can provide the volumes configurations via volumes and driver_volume_mounts and/or executor_volume_mounts arguments, and the client will take care of configuring the pod and the executors pods.

I will add an example soon, but you can check

spark-on-k8s/tests/test_spark_client.py

Lines 524 to 650 in 43fab67

	def test_submit_app_with_volumes(
	self,
	mock_create_namespaced_service,
	mock_create_namespaced_pod,
	mock_create_client,
	):
	volumes = [
	k8s.V1Volume(
	name="volume1",
	host_path=k8s.V1HostPathVolumeSource(path="/mnt/volume1"),
	),
	k8s.V1Volume(
	name="volume2",
	empty_dir=k8s.V1EmptyDirVolumeSource(medium="Memory", size_limit="1Gi"),
	),
	k8s.V1Volume(
	name="volume3",
	secret=k8s.V1SecretVolumeSource(
	secret_name="secret1", items=[k8s.V1KeyToPath(key="key1", path="path1")]
	),
	),
	k8s.V1Volume(
	name="volume4",
	config_map=k8s.V1ConfigMapVolumeSource(
	name="configmap1", items=[k8s.V1KeyToPath(key="key1", path="path1")]
	),
	),
	k8s.V1Volume(
	name="volume5",
	projected=k8s.V1ProjectedVolumeSource(
	sources=[
	k8s.V1VolumeProjection(
	secret=k8s.V1SecretProjection(items=[k8s.V1KeyToPath(key="key1", path="path1")])
	),
	k8s.V1VolumeProjection(
	config_map=k8s.V1ConfigMapProjection(
	items=[k8s.V1KeyToPath(key="key1", path="path1")]
	)
	),
	]
	),
	),
	k8s.V1Volume(
	name="volume6",
	nfs=k8s.V1NFSVolumeSource(server="nfs-server", path="/mnt/volume6", read_only=True),
	),
	k8s.V1Volume(
	name="volume7",
	persistent_volume_claim=k8s.V1PersistentVolumeClaimVolumeSource(
	claim_name="pvc1", read_only=True
	),
	),
	k8s.V1Volume(
	name="not-used-volume",
	persistent_volume_claim=k8s.V1PersistentVolumeClaimVolumeSource(
	claim_name="pvc2", read_only=True
	),
	),
	]
	driver_volume_mounts = [
	k8s.V1VolumeMount(name="volume1", mount_path="/mnt/volume1"),
	k8s.V1VolumeMount(name="volume2", mount_path="/mnt/volume2", sub_path="sub-path"),
	k8s.V1VolumeMount(name="volume3", mount_path="/mnt/volume3"),
	k8s.V1VolumeMount(name="volume4", mount_path="/mnt/volume4"),
	k8s.V1VolumeMount(name="volume5", mount_path="/mnt/volume5"),
	k8s.V1VolumeMount(name="volume6", mount_path="/mnt/volume6"),
	k8s.V1VolumeMount(name="volume7", mount_path="/mnt/volume7", read_only=True),
	]
	executor_volume_mounts = [
	k8s.V1VolumeMount(name="volume1", mount_path="/mnt/volume1"),
	k8s.V1VolumeMount(name="volume2", mount_path="/mnt/volume2", sub_path="sub-path"),
	k8s.V1VolumeMount(name="volume6", mount_path="/mnt/volume3"),
	k8s.V1VolumeMount(name="volume7", mount_path="/mnt/volume4", read_only=True),
	]
	spark_client = SparkOnK8S()
	spark_client.submit_app(
	image="pyspark-job",
	app_path="local:///opt/spark/work-dir/job.py",
	namespace="spark",
	service_account="spark",
	app_name="pyspark-job-example",
	app_arguments=["100000"],
	app_waiter="no_wait",
	image_pull_policy="Never",
	ui_reverse_proxy=True,
	driver_resources=PodResources(cpu=1, memory=2048, memory_overhead=1024),
	executor_instances=ExecutorInstances(min=2, max=5, initial=5),
	volumes=volumes,
	driver_volume_mounts=driver_volume_mounts,
	executor_volume_mounts=executor_volume_mounts,
	)
	expected_app_name = "pyspark-job-example"
	expected_app_id = f"{expected_app_name}-20240114121231"
	created_pod = mock_create_namespaced_pod.call_args[1]["body"]
	assert created_pod.metadata.labels["spark-app-name"] == expected_app_name
	assert created_pod.metadata.labels["spark-app-id"] == expected_app_id
	created_pod = mock_create_namespaced_pod.call_args[1]["body"]
	assert created_pod.spec.volumes == volumes
	assert created_pod.spec.containers[0].volume_mounts == driver_volume_mounts
	arguments = created_pod.spec.containers[0].args
	volumes_args = {
	conf.split("=")[0]: conf.split("=")[1]
	for conf in arguments
	if conf.startswith("spark.kubernetes.executor.volumes.")
	}
	assert volumes_args == {
	"spark.kubernetes.executor.volumes.hostPath.volume1.path": "/mnt/volume1",
	"spark.kubernetes.executor.volumes.emptyDir.volume2.medium": "Memory",
	"spark.kubernetes.executor.volumes.emptyDir.volume2.sizeLimit": "1Gi",
	"spark.kubernetes.executor.volumes.nfs.volume6.path": "/mnt/volume6",
	"spark.kubernetes.executor.volumes.nfs.volume6.readOnly": "true",
	"spark.kubernetes.executor.volumes.nfs.volume6.server": "nfs-server",
	"spark.kubernetes.executor.volumes.persistentVolumeClaim.volume7.claimName": "pvc1",
	"spark.kubernetes.executor.volumes.persistentVolumeClaim.volume7.readOnly": "true",
	"spark.kubernetes.executor.volumes.hostPath.volume1.mount.path": "/mnt/volume1",
	"spark.kubernetes.executor.volumes.emptyDir.volume2.mount.path": "/mnt/volume2",
	"spark.kubernetes.executor.volumes.emptyDir.volume2.mount.subPath": "sub-path",
	"spark.kubernetes.executor.volumes.nfs.volume6.mount.path": "/mnt/volume3",
	"spark.kubernetes.executor.volumes.persistentVolumeClaim.volume7.mount.path": "/mnt/volume4",
	"spark.kubernetes.executor.volumes.persistentVolumeClaim.volume7.mount.readOnly": "true",
	}

to see how this should be done (driver/executor_volume_mounts accept a value of type list[k8s.V1VolumeMount] and volumes accepts a value of type list[k8s.V1Volume], where k8s is kubernetes.client).

We can re-create the issue if it doesn't work as explained above.

[devops-eda]

ok let me check nd will update you . Thank you so much for your response .

[devops-eda]

Hi hussein,
I wiil try but not able to mount the volume.
below is the python submit file.

import sys
from spark_on_k8s.client import SparkOnK8S, ExecutorInstances, PodResources
from kubernetes import client as k8s

Define the volume for PersistentVolumeClaim

volumes = k8s.V1Volume(
name="spark-storage",
persistent_volume_claim=k8s.V1PersistentVolumeClaimVolumeSource(claim_name="cafebot-pvc")
)

Define volume mounts for driver and executor

driver_volume_mounts = k8s.V1VolumeMount(
name="spark-storage",
mount_path="/opt/cafebot"
)

executor_volume_mounts = k8s.V1VolumeMount(
name="spark-storage",
mount_path="/opt/cafebot"
)

Initialize the SparkOnK8S client

spark_client = SparkOnK8S()

Submit the Spark application

try:
spark_client.submit_app(
image="spark-submition:poc-1.7",
app_path="local:///opt/cafebot/spark-host/process_data.py",
app_arguments=[],
app_name="pyspark-job",
namespace="cafebot",
service_account="spark",
app_waiter="log",
ui_reverse_proxy=True,
driver_resources=PodResources(cpu=1, memory=512, memory_overhead=128),
executor_resources=PodResources(cpu=1, memory=512, memory_overhead=128),
executor_instances=ExecutorInstances(initial=2),
driver_volume_mounts=[driver_volume_mounts],
executor_volume_mounts=[executor_volume_mounts],
volumes=[volumes]
)
except Exception as e:
print(f"Error submitting Spark application: {e}", file=sys.stderr)

while I run this file getting this error
ERROR:

java.util.NoSuchElementException: persistentVolumeClaim.spark-storage.options.claimName is required for persistentVolumeClaim