make hadoop related options configurable in yarn_jobrunner.py eu-cdse/openeo-cdse-infra#635

Initially the values were Terrascope specific, now they can be configured via the GpsBackendConfig.

Author: JeroenVerstraelen

openeogeotrellis/config/config.py

@@ -275,6 +275,10 @@ class GpsBackendConfig(OpenEoBackendConfig):
         "/tmp_epod:/tmp_epod:rw,"
     )
     batch_user_docker_mounts: dict[str, List[str]] = {}
+    batch_spark_eventlog_dir: str = ""
+    batch_spark_history_fs_logdirectory: str = ""
+    batch_spark_yarn_historyserver_address: str = ""
+    batch_yarn_container_runtime_docker_client_config: str = ""
     gdalinfo_from_file: bool = True
     gdalinfo_python_call: bool = False
     gdalinfo_use_subprocess: bool = False

openeogeotrellis/deploy/submit_batch_job_spark3.sh

@@ -59,6 +59,10 @@ docker_mounts=${34-"/var/lib/sss/pubconf/krb5.include.d:/var/lib/sss/pubconf/krb
 udf_python_dependencies_archive_path=${35}
 propagatable_web_app_driver_envars=${36}
 python_max_memory=${37-"-1"}
+spark_eventlog_dir=${38}
+spark_history_fs_logdirectory=${39}
+spark_yarn_historyserver_address=${40}
+yarn_container_runtime_docker_client_config=${41}
 
 
 pysparkPython="/opt/venv/bin/python"
@@ -131,6 +135,13 @@ batch_job_driver_envar_arguments()
   echo "${arguments}"
 }
 
+docker_client_config_args()
+{
+  if [ -n "${yarn_container_runtime_docker_client_config}" ]; then
+    echo "--conf spark.yarn.appMasterEnv.YARN_CONTAINER_RUNTIME_DOCKER_CLIENT_CONFIG=${yarn_container_runtime_docker_client_config} --conf spark.executorEnv.YARN_CONTAINER_RUNTIME_DOCKER_CLIENT_CONFIG=${yarn_container_runtime_docker_client_config}"
+  fi
+}
+
 spark-submit \
  --master yarn --deploy-mode cluster \
  ${run_as} \
@@ -175,6 +186,7 @@ spark-submit \
  --conf spark.yarn.appMasterEnv.OPENEO_EJR_BACKEND_ID=${ejr_backend_id} \
  --conf spark.yarn.appMasterEnv.OPENEO_EJR_OIDC_CLIENT_CREDENTIALS=${ejr_oidc_client_credentials} \
  $(batch_job_driver_envar_arguments) \
+ $(docker_client_config_args) \
  --conf spark.executorEnv.AWS_REGION=${AWS_REGION} --conf spark.yarn.appMasterEnv.AWS_REGION=${AWS_REGION} \
  --conf spark.executorEnv.AWS_ACCESS_KEY_ID=${AWS_ACCESS_KEY_ID} --conf spark.yarn.appMasterEnv.AWS_ACCESS_KEY_ID=${AWS_ACCESS_KEY_ID} \
  --conf spark.executorEnv.AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY} --conf spark.yarn.appMasterEnv.AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY} \
@@ -201,14 +213,14 @@ spark-submit \
  --conf spark.hadoop.yarn.client.failover-proxy-provider=org.apache.hadoop.yarn.client.ConfiguredRMFailoverProxyProvider \
  --conf spark.shuffle.service.name=spark_shuffle_320 --conf spark.shuffle.service.port=7557 \
  --conf spark.eventLog.enabled=true \
- --conf spark.eventLog.dir=hdfs:///spark2-history/ \
- --conf spark.history.fs.logDirectory=hdfs:///spark2-history/ \
+ --conf spark.eventLog.dir=${spark_eventlog_dir} \
+ --conf spark.history.fs.logDirectory=${spark_history_fs_logdirectory} \
  --conf spark.history.kerberos.enabled=true \
  --conf spark.history.kerberos.keytab=/etc/security/keytabs/spark.service.keytab \
  --conf spark.history.kerberos.principal=spark/_HOST@VGT.VITO.BE \
  --conf spark.history.provider=org.apache.spark.deploy.history.FsHistoryProvider \
  --conf spark.history.store.path=/var/lib/spark2/shs_db \
- --conf spark.yarn.historyServer.address=epod-ha.vgt.vito.be:18481 \
+ --conf spark.yarn.historyServer.address=${spark_yarn_historyserver_address} \
  --conf spark.archives=${archives} \
  --conf spark.extraListeners=org.openeo.sparklisteners.LogErrorSparkListener,org.openeo.sparklisteners.BatchJobProgressListener \
  --conf spark.sql.adaptive.coalescePartitions.parallelismFirst=false \

openeogeotrellis/integrations/yarn_jobrunner.py

@@ -207,6 +207,18 @@ class BatchJobSubmitArgs:
     Passed to `--conf spark.executor.pyspark.memory={python_max_memory}b`.
     See https://spark.apache.org/docs/latest/configuration.html
     """
+    
+    spark_eventlog_dir: str
+    """Directory for Spark event logs"""
+    
+    spark_history_fs_logdirectory: str
+    """Directory for Spark history server logs"""
+    
+    spark_yarn_historyserver_address: str
+    """Address of Spark history server"""
+    
+    yarn_container_runtime_docker_client_config: str
+    """Path to Docker client configuration for YARN containers (optional)"""
 
     def to_args_list(self) -> list[str]:
         """Convert the dataclass to a list of string arguments for subprocess."""
@@ -249,6 +261,10 @@ def to_args_list(self) -> list[str]:
             self.udf_python_dependencies_archive_path,
             self.propagatable_web_app_driver_envars,
             self.python_max_memory,
+            self.spark_eventlog_dir,
+            self.spark_history_fs_logdirectory,
+            self.spark_yarn_historyserver_address,
+            self.yarn_container_runtime_docker_client_config,
         ]
 
 
@@ -438,6 +454,15 @@ def as_boolean_arg(job_option_key: str, default_value: str) -> str:
                 principal_value = "no_principal"
                 keytab_value = "no_keytab"
 
+            backend_config = get_backend_config()
+            
+            if not backend_config.batch_spark_eventlog_dir:
+                raise InternalException("batch_spark_eventlog_dir must be configured in backend config")
+            if not backend_config.batch_spark_history_fs_logdirectory:
+                raise InternalException("batch_spark_history_fs_logdirectory must be configured in backend config")
+            if not backend_config.batch_spark_yarn_historyserver_address:
+                raise InternalException("batch_spark_yarn_historyserver_address must be configured in backend config")
+            
             # Create structured arguments using dataclass
             submit_args = BatchJobSubmitArgs(
                 script_location=script_location,
@@ -471,13 +496,17 @@ def as_boolean_arg(job_option_key: str, default_value: str) -> str:
                 log_level=options.log_level,
                 openeo_backend_config=os.environ.get(ConfigGetter.OPENEO_BACKEND_CONFIG, ""),
                 udf_python_dependencies_folder_path=str(job_work_dir / UDF_PYTHON_DEPENDENCIES_FOLDER_NAME),
-                ejr_api=get_backend_config().ejr_api or "",
-                ejr_backend_id=get_backend_config().ejr_backend_id,
+                ejr_api=backend_config.ejr_api or "",
+                ejr_backend_id=backend_config.ejr_backend_id,
                 ejr_oidc_client_credentials=os.environ.get("OPENEO_EJR_OIDC_CLIENT_CREDENTIALS", ""),
                 docker_mounts=docker_mounts,
                 udf_python_dependencies_archive_path=str(job_work_dir / UDF_PYTHON_DEPENDENCIES_ARCHIVE_NAME),
                 propagatable_web_app_driver_envars=os.environ.get("OPENEO_PROPAGATABLE_WEB_APP_DRIVER_ENVARS", ""),
                 python_max_memory=str(byte_string_as(options.python_memory)) if options.python_memory else "-1",
+                spark_eventlog_dir=backend_config.batch_spark_eventlog_dir,
+                spark_history_fs_logdirectory=backend_config.batch_spark_history_fs_logdirectory,
+                spark_yarn_historyserver_address=backend_config.batch_spark_yarn_historyserver_address,
+                yarn_container_runtime_docker_client_config=backend_config.batch_yarn_container_runtime_docker_client_config,
             )
             args = submit_args.to_args_list()