Repository: AbsaOSS/pramen (language: python)

Issue:
Allow to configure Spark configuration for metatables
## Background
Sometimes it is needed to use 'magic' committer for some tables in the metastore, but to use default Spark configuration to write to other tables.

It could be helpful if special Spark configuration can be specified for individual tables in the metastore.

The configuration should be restored after the write.

## Feature
Allow to configure Spark configuration for metatables.

## Example
```hocon
pramen.metastore {
  tables = [
    {
      name = "my_table1"
      format = "parquet"
      path = "s3://bucket1/path1"
    },
    {
      name = "my_table2"
      format = "parquet"
      path = "s3a://bucket2/path2"
      spark.conf {
          spark.sql.sources.commitProtocolClass = "org.apache.spark.internal.io.cloud.PathOutputCommitProtocol"
          spark.sql.parquet.output.committer.class = "org.apache.spark.internal.io.cloud.BindingParquetOutputCommitter"
      }
    }
  ]
}
```
## Proposed Solution
- Add the ability to specify the config
- Make sure the Spark configuration is restored after the write



Discussion / hints:

Fix patch:
diff --git a/README.md b/README.md
index 64b536aa0..16e73f931 100644
--- a/README.md
+++ b/README.md
@@ -328,6 +328,13 @@ pramen.metastore {
       # Alternatively, you can set the time window in days that from the current system
       # date that the table could be written to. This is helpful is the table is archived.
       #information.date.max.days.behind = 180
+
+      # Optional Spark configuration that will be used when writing to the table
+      # Useful to use Spark Committers (partitioned, directory, magic) only for some of tables. 
+      spark.conf {
+        spark.sql.sources.commitProtocolClass = "org.apache.spark.internal.io.cloud.PathOutputCommitProtocol"
+        spark.sql.parquet.output.committer.class = "org.apache.spark.internal.io.cloud.BindingParquetOutputCommitter"
+      }
     }
   ]
 }
@@ -2108,7 +2115,7 @@ and 'python.class' to refer to the transformer.
   
   # Arbitrary Spark configuration
   # You can use any configuration option from the official documentation: https://spark.apache.org/docs/latest/configuration.html
-  spark.config {
+  spark.conf {
     spark.executor.instances = 4
     spark.executor.cores = 1
     spark.executor.memory = "4g"
@@ -2816,11 +2823,16 @@ Hive configuration and query templates can be defined globally:
 ```hocon
 pramen {
   # Hive configuration for the Spark metastore
-  spark.conf.option = {
+  spark.conf = {
     hive.metastore.uris = "thrift://host1:9083,thrift://host2:9083"
     spark.sql.warehouse.dir = "/hive/warehouse"
   }
   
+  hadoop.conf = {
+    # Any Hadoop custom options go here
+    fs.s3a.path.style.access = true
+  }
+  
   hive {
     # The API to use to query Hive. Valid values are: "sql" (default), "spark_catalog"
     hive.api = "sql"
diff --git a/pramen/core/src/main/resources/reference.conf b/pramen/core/src/main/resources/reference.conf
index 7f5f48c95..578abc155 100644
--- a/pramen/core/src/main/resources/reference.conf
+++ b/pramen/core/src/main/resources/reference.conf
@@ -108,7 +108,7 @@ pramen {
   # If this is set the current date will overridden by the specified value.
   #current.date =
 
-  spark.conf.option = {
+  #spark.conf = {
     # Pass arbitrary Spark Configuration when initializing Spark Session
 
     # For example, alternative way of writing legacy parquet will be
@@ -117,10 +117,10 @@ pramen {
     # These options are needed so the job could create Hive tables
     #hive.metastore.uris = "thrift://host1:9083,thrift://host2:9083"
     #spark.sql.warehouse.dir = "/hive/warehouse"
-  }
+  #}
 
   # Other option(s) might be
-  # spark.conf.option.spark.sql.parquet.binaryAsString = true
+  # spark.conf.spark.sql.parquet.binaryAsString = true
 
   # Default infroamation date expression for daily jobs
   default.daily.output.info.date.expr = "@runDate"
@@ -201,7 +201,7 @@ mail {
 hadoop.redacted.tokens = [ password, secret, session.token, access.key ]
 
 # Hadoop options to access S3
-# hadoop.option {
+# hadoop.conf {
 #   # Authentication provider. Can be
 #   # * org.apache.hadoop.fs.s3a.TemporaryAWSCredentialsProvider
 #   # * com.amazonaws.auth.profile.ProfileCredentialsProvider
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/RunnerCommons.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/RunnerCommons.scala
index 3dfa8a3e1..6266bf92e 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/RunnerCommons.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/RunnerCommons.scala
@@ -44,7 +44,7 @@ object RunnerCommons {
     * Such options are:
     * - pramen.exit.code.enabled
     * - JVM security options: javax.*, java.* (see JavaXConfig.setJavaXProperties)
-    * - Spark Session config: pramen.spark.conf.option.*
+    * - Spark Session config: pramen.spark.conf.*
     *   This is because restarting Spark is not supported by most of runtimes.
     *
     * On the other hand, Hadoop options (hadoop.option.*), will be applied for each workflow.
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/config/Keys.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/config/Keys.scala
index fd56d328e..6b5562048 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/config/Keys.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/config/Keys.scala
@@ -31,8 +31,10 @@ object Keys {
 
   val HADOOP_REDACT_TOKENS = "hadoop.redacted.tokens"
   val HADOOP_OPTION_PREFIX = "hadoop.option"
+  val HADOOP_OPTION_PREFIX_V2 = "hadoop.conf"
 
   val EXTRA_OPTIONS_PREFIX = "pramen.spark.conf.option"
+  val EXTRA_OPTIONS_PREFIX_V2 = "pramen.spark.conf"
 
   val ENABLE_HIVE_SUPPORT = "pramen.enable.hive"
 
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/MetastoreImpl.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/MetastoreImpl.scala
index 0344ef4ef..5b5651372 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/MetastoreImpl.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/MetastoreImpl.scala
@@ -26,6 +26,7 @@ import za.co.absa.pramen.core.app.config.InfoDateConfig
 import za.co.absa.pramen.core.app.config.InfoDateConfig.DEFAULT_DATE_FORMAT
 import za.co.absa.pramen.core.app.config.RuntimeConfig.UNDERCOVER
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
+import za.co.absa.pramen.core.config.Keys
 import za.co.absa.pramen.core.metastore.model.MetaTable
 import za.co.absa.pramen.core.metastore.peristence.{MetastorePersistence, TransientJobManager}
 import za.co.absa.pramen.core.utils.ConfigUtils
@@ -38,6 +39,8 @@ class MetastoreImpl(appConfig: Config,
                     bookkeeper: Bookkeeper,
                     metadata: MetadataManager,
                     skipBookKeepingUpdates: Boolean)(implicit spark: SparkSession) extends Metastore {
+  import MetastoreImpl._
+
   private val log = LoggerFactory.getLogger(this.getClass)
 
   override def getRegisteredTables: Seq[String] = tableDefs.map(_.name)
@@ -92,7 +95,13 @@ class MetastoreImpl(appConfig: Config,
     val mt = getTableDef(tableName)
     val isTransient = mt.format.isTransient
     val start = Instant.now.getEpochSecond
-    val stats = MetastorePersistence.fromMetaTable(mt, appConfig).saveTable(infoDate, df, inputRecordCount)
+
+    var stats = MetaTableStats(0, None)
+
+    withSparkConfig(mt.sparkConfig) {
+      stats = MetastorePersistence.fromMetaTable(mt, appConfig).saveTable(infoDate, df, inputRecordCount)
+    }
+
     val finish = Instant.now.getEpochSecond
 
     if (!skipBookKeepingUpdates) {
@@ -236,6 +245,8 @@ class MetastoreImpl(appConfig: Config,
 }
 
 object MetastoreImpl {
+  private val log = LoggerFactory.getLogger(this.getClass)
+
   val METASTORE_KEY = "pramen.metastore.tables"
   val DEFAULT_RECORDS_PER_PARTITION = 500000
 
@@ -249,5 +260,36 @@ object MetastoreImpl {
 
     new MetastoreImpl(conf, tableDefs, bookkeeper, metadataManager, isUndercover)
   }
+
+  private[core] def withSparkConfig(sparkConfig: Map[String, String])
+                                   (action: => Unit)
+                                   (implicit spark: SparkSession): Unit = {
+    val savedConfig = sparkConfig.map {
+      case (k, _) => (k, spark.conf.getOption(k))
+    }
+
+    sparkConfig.foreach {
+      case (k, v) =>
+        val redactedValue = ConfigUtils.renderRedactedKeyValue(k, v, Keys.KEYS_TO_REDACT)
+        log.info(s"Setting $redactedValue...")
+        spark.conf.set(k, v)
+    }
+
+    try {
+      action
+    } finally {
+      savedConfig.foreach {
+        case (k, opt) => opt match {
+          case Some(v) =>
+            val redactedValue = ConfigUtils.renderRedactedKeyValue(k, v, Keys.KEYS_TO_REDACT)
+            log.info(s"Restoring $redactedValue...")
+            spark.conf.set(k, v)
+          case None =>
+            log.info(s"Unsetting '$k'...")
+            spark.conf.unset(k)
+        }
+      }
+    }
+  }
 }
 
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/HiveDefaultConfig.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/HiveDefaultConfig.scala
index 169223bb8..018f52a7f 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/HiveDefaultConfig.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/HiveDefaultConfig.scala
@@ -27,7 +27,7 @@ import za.co.absa.pramen.core.utils.hive.HiveQueryTemplates
   * It works like this. The pipeline can define a global Hive configuration and query templates like this:
   * {{{
   * pramen {
-  *   spark.conf.option = {
+  *   spark.conf = {
   *     hive.metastore.uris = "thrift://host1:9083,thrift://host2:9083"
   *     spark.sql.warehouse.dir = "/hive/warehouse"
   *   }
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/MetaTable.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/MetaTable.scala
index 3a3894c13..3e57709b9 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/MetaTable.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/metastore/model/MetaTable.scala
@@ -44,6 +44,7 @@ import scala.util.{Failure, Success, Try}
   * @param trackDaysExplicitlySet if true, trackDays was set explicitly. If false, trackDays is taken from workflow defaults.
   * @param readOptions            The read options for the table.
   * @param writeOptions           The write options for the table.
+  * @param sparkConfig            Special Spark configuration to use when writing to the table.
   */
 case class MetaTable(
                       name: String,
@@ -60,7 +61,8 @@ case class MetaTable(
                       trackDays: Int,
                       trackDaysExplicitlySet: Boolean,
                       readOptions: Map[String, String],
-                      writeOptions: Map[String, String]
+                      writeOptions: Map[String, String],
+                      sparkConfig: Map[String, String]
                     )
 
 object MetaTable {
@@ -76,6 +78,7 @@ object MetaTable {
   val WRITE_OPTION_KEY = "write.option"
   val TABLE_HIVE_CONFIG_PREFIX = "hive"
   val DEFAULT_HIVE_CONFIG_PREFIX = "pramen.hive"
+  val SPARK_CONFIG_PREFIX = "spark.conf"
 
   def fromConfig(conf: Config, infoDateConfig: InfoDateConfig, key: String): Seq[MetaTable] = {
     val defaultInfoDateColumnName = infoDateConfig.columnName
@@ -139,6 +142,7 @@ object MetaTable {
 
     val readOptions = ConfigUtils.getExtraOptions(conf, READ_OPTION_KEY)
     val writeOptions = ConfigUtils.getExtraOptions(conf, WRITE_OPTION_KEY)
+    val sparkConfig = ConfigUtils.getExtraOptions(conf, SPARK_CONFIG_PREFIX)
 
     MetaTable(name,
       description,
@@ -154,7 +158,8 @@ object MetaTable {
       trackDays,
       trackDaysExplicitlySet,
       readOptions,
-      writeOptions)
+      writeOptions,
+      sparkConfig)
   }
 
   def getMetaTableDef(table: MetaTable): MetaTableDef = {
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/OperationDef.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/OperationDef.scala
index ae43bd452..fe9f1632d 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/OperationDef.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/OperationDef.scala
@@ -69,6 +69,7 @@ object OperationDef {
   val FILTERS_KEY = "filters"
   val NOTIFICATION_TARGETS_KEY = "notification.targets"
   val SPARK_CONFIG_PREFIX = "spark.config"
+  val SPARK_CONFIG_PREFIX_V2 = "spark.conf"
   val EXTRA_OPTIONS_PREFIX = "option"
 
   val DEFAULT_CONSUME_THREADS = 1
@@ -98,9 +99,13 @@ object OperationDef {
     val schemaTransformations = TransformExpression.fromConfig(conf, SCHEMA_TRANSFORMATIONS_KEY, parent)
     val filters = ConfigUtils.getOptListStrings(conf, FILTERS_KEY)
     val notificationTargets = ConfigUtils.getOptListStrings(conf, NOTIFICATION_TARGETS_KEY)
-    val sparkConfigOptions = ConfigUtils.getExtraOptions(conf, SPARK_CONFIG_PREFIX)
+    val sparkConfigOptions = ConfigUtils.getExtraOptions(conf, SPARK_CONFIG_PREFIX) ++ ConfigUtils.getExtraOptions(conf, SPARK_CONFIG_PREFIX_V2)
     val extraOptions = ConfigUtils.getExtraOptions(conf, EXTRA_OPTIONS_PREFIX)
 
+    if (conf.hasPath(SPARK_CONFIG_PREFIX)) {
+      log.warn(s"Using legacy '$SPARK_CONFIG_PREFIX' option. Please, use the new option: '$SPARK_CONFIG_PREFIX_V2'")
+    }
+
     val outputInfoDateExpression = outputInfoDateExpressionOpt match {
       case Some(expr) => expr
       case None       =>
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransferTable.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransferTable.scala
index 600048be2..b3058648e 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransferTable.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/pipeline/TransferTable.scala
@@ -23,7 +23,7 @@ import za.co.absa.pramen.core.app.config.InfoDateConfig
 import za.co.absa.pramen.core.config.InfoDateOverride
 import za.co.absa.pramen.core.metastore.model.{HiveConfig, MetaTable}
 import za.co.absa.pramen.core.model.QueryBuilder
-import za.co.absa.pramen.core.pipeline.OperationDef.WARN_MAXIMUM_EXECUTION_TIME_SECONDS_KEY
+import za.co.absa.pramen.core.pipeline.OperationDef.{SPARK_CONFIG_PREFIX, WARN_MAXIMUM_EXECUTION_TIME_SECONDS_KEY}
 import za.co.absa.pramen.core.utils.{AlgorithmUtils, ConfigUtils}
 
 import java.time.LocalDate
@@ -44,6 +44,7 @@ case class TransferTable(
                           columns: Seq[String],
                           readOptions: Map[String, String],
                           writeOptions: Map[String, String],
+                          sparkConfig: Map[String, String],
                           sourceOverrideConf: Option[Config],
                           sinkOverrideConf: Option[Config]
                         ) {
@@ -56,7 +57,7 @@ case class TransferTable(
   }
 
   def getMetaTable: MetaTable = {
-    MetaTable(jobMetaTableName, "", DataFormat.Null(), "", "", HiveConfig.getNullConfig, None, None, hivePreferAddPartition = true, None, infoDateStart, trackDays, trackDaysExplicitlySet = trackDaysExplicitlySet, readOptions, writeOptions)
+    MetaTable(jobMetaTableName, "", DataFormat.Null(), "", "", HiveConfig.getNullConfig, None, None, hivePreferAddPartition = true, None, infoDateStart, trackDays, trackDaysExplicitlySet = trackDaysExplicitlySet, readOptions, writeOptions, sparkConfig)
   }
 }
 
@@ -86,6 +87,7 @@ object TransferTable {
     val filters = ConfigUtils.getOptListStrings(conf, FILTERS_KEY)
     val readOptions = ConfigUtils.getExtraOptions(conf, "read.option")
     val writeOptions = ConfigUtils.getExtraOptions(conf, "output")
+    val sparkConfig = ConfigUtils.getExtraOptions(conf, SPARK_CONFIG_PREFIX)
 
     val outputMetaTableName = jobMetaTableOpt.getOrElse(s"$query -> $sinkName")
 
@@ -107,7 +109,7 @@ object TransferTable {
     val startDate = infoDateOverride.startDate.getOrElse(defaultStartDate)
     val jobMetaTable = getOutputTableName(jobMetaTableOpt, query, sinkName)
 
-    TransferTable(query, jobMetaTable, conf, dateFromExpr, dateToExpr, startDate, trackDays, trackDaysExplicitlySet, maximumExecutionTimeSeconds, transformations, filters, columns, readOptions, writeOptions, sourceOverrideConf, sinkOverrideConf)
+    TransferTable(query, jobMetaTable, conf, dateFromExpr, dateToExpr, startDate, trackDays, trackDaysExplicitlySet, maximumExecutionTimeSeconds, transformations, filters, columns, readOptions, writeOptions, sparkConfig, sourceOverrideConf, sinkOverrideConf)
   }
 
   def fromConfig(conf: Config, infoDateConfig: InfoDateConfig, arrayPath: String, sinkName: String): Seq[TransferTable] = {
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/JdbcUrlSelectorImpl.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/JdbcUrlSelectorImpl.scala
index 16fe0559e..32e7b44c1 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/JdbcUrlSelectorImpl.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/reader/JdbcUrlSelectorImpl.scala
@@ -59,7 +59,7 @@ class JdbcUrlSelectorImpl(val jdbcConfig: JdbcConfig) extends JdbcUrlSelector{
 
     jdbcConfig.database.foreach(d  => log.info(s"Database:           $d"))
     jdbcConfig.user.foreach(user   => log.info(s"User:               $user"))
-    jdbcConfig.password.foreach(_  => log.info(s"Password:           [redacted]"))
+    jdbcConfig.password.filter(_.nonEmpty).foreach(_  => log.info(s"Password:           [redacted]"))
     jdbcConfig.fetchSize.foreach(n => log.info(s"Fetch size:         $n"))
 
     log.info(s"Auto commit:        ${jdbcConfig.autoCommit}")
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/AppRunner.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/AppRunner.scala
index e65faafc2..20faa72b0 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/AppRunner.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/AppRunner.scala
@@ -125,7 +125,7 @@ object AppRunner {
   }
 
   private[core] def getSparkSession(implicit conf: Config,
-                                         state: PipelineState): Try[SparkSession] = {
+                                    state: PipelineState): Try[SparkSession] = {
     handleFailure(Try {
       PipelineSparkSessionBuilder.buildSparkSession(conf)
     }, state, "Spark Session creation")
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/PipelineSparkSessionBuilder.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/PipelineSparkSessionBuilder.scala
index 9ac460607..f0d6f4416 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/PipelineSparkSessionBuilder.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/runner/PipelineSparkSessionBuilder.scala
@@ -34,7 +34,7 @@ object PipelineSparkSessionBuilder {
     *
     * Extra options can be passed as
     * {{{
-    *   pramen.spark.conf.option {
+    *   pramen.spark.conf {
     *     spark.config.option = "value"
     *   }
     * }}}
@@ -50,7 +50,16 @@ object PipelineSparkSessionBuilder {
     val isHiveEnabled = conf.getBoolean(ENABLE_HIVE_SUPPORT)
     log.info(s"Hive support enabled = $isHiveEnabled")
 
-    val extraOptions = ConfigUtils.getExtraOptions(conf, EXTRA_OPTIONS_PREFIX)
+    val extraOptionsLegacy = ConfigUtils.getExtraOptions(conf, EXTRA_OPTIONS_PREFIX)
+    val extraOptionsNew = ConfigUtils.getExtraOptions(conf, EXTRA_OPTIONS_PREFIX_V2)
+
+    val extraOptions = if (extraOptionsLegacy.nonEmpty) {
+      log.warn(s"Using legacy '$EXTRA_OPTIONS_PREFIX' option. Please, use the new option: '$EXTRA_OPTIONS_PREFIX_V2'")
+      extraOptionsLegacy
+    } else {
+      extraOptionsNew
+    }
+
     log.info("Extra Spark Config:")
     ConfigUtils.renderExtraOptions(extraOptions, KEYS_TO_REDACT)(s => log.info(s))
 
@@ -83,12 +92,15 @@ object PipelineSparkSessionBuilder {
   def applyHadoopConfig(spark: SparkSession, conf: Config): SparkSession = {
     val redactTokens = ConfigUtils.getOptListStrings(conf, HADOOP_REDACT_TOKENS).toSet
 
-    if (conf.hasPath(HADOOP_OPTION_PREFIX)) {
+    if (conf.hasPath(HADOOP_OPTION_PREFIX) || conf.hasPath(HADOOP_OPTION_PREFIX_V2)) {
+      if (conf.hasPath(HADOOP_OPTION_PREFIX)) {
+        log.warn(s"Using legacy '$HADOOP_OPTION_PREFIX' option. Please, use the new option: '$HADOOP_OPTION_PREFIX_V2'")
+      }
       val sc = spark.sparkContext
-      val hadoopOptions = ConfigUtils.getExtraOptions(conf, HADOOP_OPTION_PREFIX)
+      val hadoopOptions = ConfigUtils.getExtraOptions(conf, HADOOP_OPTION_PREFIX) ++ ConfigUtils.getExtraOptions(conf, HADOOP_OPTION_PREFIX_V2)
 
       hadoopOptions.foreach { case (key, value) =>
-        val redactedValue = ConfigUtils.getRedactedValue(key, value, redactTokens).toString
+        val redactedValue = ConfigUtils.renderRedactedKeyValue(key, value, redactTokens).toString
         log.info(s"Hadoop config: $key = $redactedValue")
 
         sc.hadoopConfiguration.set(key, value)
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/ConfigUtils.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/ConfigUtils.scala
index 349c15af2..25e75a4e1 100644
--- a/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/ConfigUtils.scala
+++ b/pramen/core/src/main/scala/za/co/absa/pramen/core/utils/ConfigUtils.scala
@@ -249,18 +249,25 @@ object ConfigUtils {
                          redactedWords: Set[String] = Set.empty[String])
                         (action: String => Unit): Unit = {
     if (extraOptions.nonEmpty) {
-      val p = "\""
+
       extraOptions.foreach { case (key, value) =>
-        val lowerCaseKey = key.toLowerCase()
-        if (redactedWords.exists(word => lowerCaseKey.contains(word))) {
-          action(s"$key = [redacted]")
-        } else {
-          action(s"$key = $p$value$p")
-        }
+        val v = renderRedactedKeyValue(key, value, redactedWords)
+
+        action(v)
       }
     }
   }
 
+  def renderRedactedKeyValue(key: String, value: String, redactedWords: Set[String]): String = {
+    val p = "\""
+    val lowerCaseKey = key.toLowerCase()
+    if (redactedWords.exists(word => lowerCaseKey.contains(word))) {
+      s"$key = [redacted]"
+    } else {
+      s"$key = $p$value$p"
+    }
+  }
+
   /**
     * Given a configuration returns a new configuration which has all sensitive keys redacted.
     *
diff --git a/pramen/examples/combined_example.sh b/pramen/examples/combined_example.sh
index 1aeb2aa48..634348765 100755
--- a/pramen/examples/combined_example.sh
+++ b/pramen/examples/combined_example.sh
@@ -67,7 +67,7 @@ mail {
 }
 
 # This is to run properly in local mode
-spark.conf.option {
+spark.conf {
     spark.ui.enabled = "false"
     spark.driver.bindAddress = "127.0.0.1"
     spark.driver.host = "127.0.0.1"


Test patch:
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/MetastoreSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/MetastoreSuite.scala
index ae84235da..079470b5f 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/MetastoreSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/MetastoreSuite.scala
@@ -471,6 +471,37 @@ class MetastoreSuite extends AnyWordSpec with SparkTestBase with TextComparisonF
     }
   }
 
+  "withSparkConfig()" should {
+    "set the config at runtime, and restore the original config afterwards" in {
+      val sparkConfig = Map(
+        "spark.sql.sources.commitProtocolClass" -> "org.apache.spark.internal.io.cloud.PathOutputCommitProtocol",
+        "spark.sql.parquet.output.committer.class" -> "org.apache.spark.internal.io.cloud.BindingParquetOutputCommitter",
+        "spark.pramen.test" -> "test"
+      )
+
+      var inner1: String = null
+      var inner2: String = null
+      var inner3: String = null
+
+      MetastoreImpl.withSparkConfig(sparkConfig) {
+        inner1 = spark.conf.get("spark.sql.sources.commitProtocolClass")
+        inner2 = spark.conf.get("spark.sql.parquet.output.committer.class")
+        inner3 = spark.conf.get("spark.pramen.test")
+      }
+
+      val outer1 = spark.conf.get("spark.sql.sources.commitProtocolClass")
+      val outer2 = spark.conf.get("spark.sql.parquet.output.committer.class")
+      val outer3 = spark.conf.getOption("spark.pramen.test")
+
+      assert(inner1 == "org.apache.spark.internal.io.cloud.PathOutputCommitProtocol")
+      assert(inner2 == "org.apache.spark.internal.io.cloud.BindingParquetOutputCommitter")
+      assert(inner3 == "test")
+      assert(outer1 != "org.apache.spark.internal.io.cloud.PathOutputCommitProtocol")
+      assert(outer2 != "org.apache.spark.internal.io.cloud.BindingParquetOutputCommitter")
+      assert(outer3.isEmpty)
+    }
+  }
+
   def getDf: DataFrame = {
     import spark.implicits._
 
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/model/MetaTableSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/model/MetaTableSuite.scala
index 188184901..b79714bb7 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/model/MetaTableSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/metastore/model/MetaTableSuite.scala
@@ -150,6 +150,9 @@ class MetaTableSuite extends AnyWordSpec {
           |format = delta
           |path = /a/b/c
           |records.per.partition = 100
+          |spark.conf = {
+          |  key1 = value1
+          |}
           |""".stripMargin)
 
       val defaultHiveConfig = HiveDefaultConfig.getNullConfig
@@ -169,6 +172,7 @@ class MetaTableSuite extends AnyWordSpec {
       assert(metaTable.infoDateColumn == "INFO_DATE")
       assert(metaTable.infoDateFormat == "dd-MM-yyyy")
       assert(metaTable.infoDateStart.toString == "2020-01-31")
+      assert(metaTable.sparkConfig("key1") == "value1")
     }
 
     "load a metatable definition with hive table defined" in {
@@ -190,6 +194,9 @@ class MetaTableSuite extends AnyWordSpec {
           |  x = test2
           |  y = 101
           |}
+          |spark.conf = {
+          |  key1 = value1
+          |}
           |""".stripMargin)
 
       val defaultHiveConfig = HiveDefaultConfig(HiveApi.Sql,
@@ -225,6 +232,7 @@ class MetaTableSuite extends AnyWordSpec {
       assert(metaTable.readOptions("some.option.b") == "12")
       assert(metaTable.writeOptions("x") == "test2")
       assert(metaTable.writeOptions("y") == "101")
+      assert(metaTable.sparkConfig("key1") == "value1")
     }
 
     "load a metatable definition with hive overrides" in {
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/MetaTableFactory.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/MetaTableFactory.scala
index 3fbfd579b..218c81f03 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/MetaTableFactory.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/MetaTableFactory.scala
@@ -36,7 +36,8 @@ object MetaTableFactory {
                         trackDays: Int = 0,
                         trackDaysExplicitlySet: Boolean = false,
                         readOptions: Map[String, String] = Map.empty[String, String],
-                        writeOptions: Map[String, String] = Map.empty[String, String]
+                        writeOptions: Map[String, String] = Map.empty[String, String],
+                        sparkConfig: Map[String, String] = Map.empty[String, String]
                        ): MetaTable = {
     MetaTable(name,
       description,
@@ -52,6 +53,7 @@ object MetaTableFactory {
       trackDays,
       trackDaysExplicitlySet,
       readOptions,
-      writeOptions)
+      writeOptions,
+      sparkConfig)
   }
 }
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/TransferTableFactory.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/TransferTableFactory.scala
index 4199932f3..3f5bd4b23 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/TransferTableFactory.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/TransferTableFactory.scala
@@ -37,6 +37,7 @@ object TransferTableFactory {
                             columns: Seq[String] = Nil,
                             readOptions: Map[String, String] = Map.empty[String, String],
                             writeOptions: Map[String, String] = Map.empty[String, String],
+                            sparkConfig: Map[String, String] = Map.empty[String, String],
                             sourceOverrideConf: Option[Config] = None,
                             sinkOverrideConf: Option[Config] = None): TransferTable = {
     TransferTable(query,
@@ -53,6 +54,7 @@ object TransferTableFactory {
       columns,
       readOptions,
       writeOptions,
+      sparkConfig,
       sourceOverrideConf,
       sinkOverrideConf
     )
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/TransferTableSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/TransferTableSuite.scala
index a4be73520..abffbc552 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/TransferTableSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/pipeline/TransferTableSuite.scala
@@ -216,6 +216,9 @@ class TransferTableSuite extends AnyWordSpec {
         |    read.option {
         |      mergeSchema = "true"
         |    }
+        |    spark.config = {
+        |      key1 = value1
+        |    }
         |    track.days = 10
         |    information.date.start = "2022-01-19"
         |    source {
@@ -236,6 +239,7 @@ class TransferTableSuite extends AnyWordSpec {
 
         val sourceTable = tables.head.getSourceTable
 
+        assert(tables.head.sparkConfig("key1") == "value1")
         assert(sourceTable.metaTableName == "table1->mysink")
         assert(sourceTable.query.isInstanceOf[Query.Table])
         assert(sourceTable.query.asInstanceOf[Query.Table].dbTable == "table1")
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/PipelineSparkSessionBuilderSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/PipelineSparkSessionBuilderSuite.scala
index 06e9b3a23..00fb8b670 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/PipelineSparkSessionBuilderSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/runner/PipelineSparkSessionBuilderSuite.scala
@@ -74,6 +74,44 @@ class PipelineSparkSessionBuilderSuite extends AnyWordSpec {
       assert(key1Set)
       assert(key2Set)
     }
+
+    "apply the given config to the spark session version 2" in {
+      val conf = ConfigFactory.parseString(
+        s"""$HADOOP_REDACT_TOKENS = [ secret ]
+           |$HADOOP_OPTION_PREFIX_V2 {
+           |  fs.s3a.aws.credentials.provider = "com.amazonaws.auth.DefaultAWSCredentialsProviderChain"
+           |  fs.s3a.secret.key = "acs123"
+           |}
+           |""".stripMargin
+      )
+
+      val sparkMock = mock(classOf[SparkSession])
+      val scMock = mock(classOf[SparkContext])
+      val hcMock = mock(classOf[Configuration])
+
+      whenMock(sparkMock.sparkContext).thenReturn(scMock)
+      whenMock(scMock.hadoopConfiguration).thenReturn(hcMock)
+
+      var key1Set = false
+      var key2Set = false
+
+      doAnswer(new Answer[Unit] {
+        override def answer(invocationOnMock: InvocationOnMock): Unit = {
+          key1Set = true
+        }
+      }).when(hcMock).set("fs.s3a.aws.credentials.provider", "com.amazonaws.auth.DefaultAWSCredentialsProviderChain")
+
+      doAnswer(new Answer[Unit] {
+        override def answer(invocationOnMock: InvocationOnMock): Unit = {
+          key2Set = true
+        }
+      }).when(hcMock).set("fs.s3a.secret.key", "acs123")
+
+      PipelineSparkSessionBuilder.applyHadoopConfig(sparkMock, conf)
+
+      assert(key1Set)
+      assert(key2Set)
+    }
   }
 
   "getSparkAppName" should {
diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/ConfigUtilsSuite.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/ConfigUtilsSuite.scala
index 88324a5a1..fc079e66e 100644
--- a/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/ConfigUtilsSuite.scala
+++ b/pramen/core/src/test/scala/za/co/absa/pramen/core/tests/utils/ConfigUtilsSuite.scala
@@ -329,6 +329,17 @@ class ConfigUtilsSuite extends AnyWordSpec with TempDirFixture with TextComparis
     }
   }
 
+  "renderRedactedKeyValue()" should {
+    "redact keys containing the list of tokens" in {
+      val tokens = Set("secret", "password", "session.token")
+
+      assert(ConfigUtils.renderRedactedKeyValue("mytest.password", "pwd", tokens) == "mytest.password = [redacted]")
+      assert(ConfigUtils.renderRedactedKeyValue("mytest.secret", "pwd", tokens) == "mytest.secret = [redacted]")
+      assert(ConfigUtils.renderRedactedKeyValue("mytest.session.token", "pwd", tokens) == "mytest.session.token = [redacted]")
+      assert(ConfigUtils.renderRedactedKeyValue("mytest.session.name", "name", tokens) == "mytest.session.name = \"name\"")
+    }
+  }
+
   "convertToMap()" should {
     "convert a simple config" in {
       val conf = ConfigFactory.parseString(