From d12552adab9764eafec0264137aa57830ea9668d Mon Sep 17 00:00:00 2001
From: Cristian Goina <goinac@hhmi.org>
Date: Tue, 15 Apr 2025 16:17:09 -0400
Subject: [PATCH 1/8] add spark config to start spark

---
 conf/spot-extraction-modules.config           |   1 +
 modules.json                                  |   4 +-
 .../dumpsoftwareversions/environment.yml      |   2 +-
 .../custom/dumpsoftwareversions/main.nf       |  13 +-
 .../tests/main.nf.test.snap                   |  16 +-
 .../dumpsoftwareversions/tests/tags.yml       |   2 -
 subworkflows/janelia/spark_start/main.nf      | 209 +++++++++++++-----
 subworkflows/local/multiscale.nf              |   3 +-
 subworkflows/local/stitching.nf               |   3 +-
 workflows/registration.nf                     |   1 +
 workflows/spot_extraction.nf                  |   1 +
 11 files changed, 184 insertions(+), 71 deletions(-)
 delete mode 100644 modules/nf-core/custom/dumpsoftwareversions/tests/tags.yml

diff --git a/conf/spot-extraction-modules.config b/conf/spot-extraction-modules.config
index 7f509b90..1ee169a5 100644
--- a/conf/spot-extraction-modules.config
+++ b/conf/spot-extraction-modules.config
@@ -30,6 +30,7 @@ process {
 
     withName: "(.*)?RS_FISH" {
         errorStrategy = 'terminate'
+        ext.container = 'ghcr.io/janeliascicomp/rs-fish-spark:4d1b901'
         ext.args = [
             "--minIntensity=${params.rsfish_min_intensity}",
             "--maxIntensity=${params.rsfish_max_intensity}",
diff --git a/modules.json b/modules.json
index 3b4a72ab..d7cc8937 100644
--- a/modules.json
+++ b/modules.json
@@ -66,7 +66,7 @@
                     },
                     "spark_start": {
                         "branch": "main",
-                        "git_sha": "ce27f00338affaed60cbb37b4dc060f145297c23",
+                        "git_sha": "cfc0afa06f34e7c94d875bdbafb45889d9405cd2",
                         "installed_by": ["subworkflows"]
                     },
                     "spark_stop": {
@@ -82,7 +82,7 @@
                 "nf-core": {
                     "custom/dumpsoftwareversions": {
                         "branch": "master",
-                        "git_sha": "81880787133db07d9b4c1febd152c090eb8325dc",
+                        "git_sha": "05954dab2ff481bcb999f24455da29a5828af08d",
                         "installed_by": ["modules"]
                     }
                 }
diff --git a/modules/nf-core/custom/dumpsoftwareversions/environment.yml b/modules/nf-core/custom/dumpsoftwareversions/environment.yml
index 70ee54e3..c3b3413f 100644
--- a/modules/nf-core/custom/dumpsoftwareversions/environment.yml
+++ b/modules/nf-core/custom/dumpsoftwareversions/environment.yml
@@ -4,4 +4,4 @@ channels:
   - conda-forge
   - bioconda
 dependencies:
-  - bioconda::multiqc=1.20
+  - bioconda::multiqc=1.27
diff --git a/modules/nf-core/custom/dumpsoftwareversions/main.nf b/modules/nf-core/custom/dumpsoftwareversions/main.nf
index 105f9265..dd6e210d 100644
--- a/modules/nf-core/custom/dumpsoftwareversions/main.nf
+++ b/modules/nf-core/custom/dumpsoftwareversions/main.nf
@@ -1,11 +1,19 @@
+def deprecation_message = """
+WARNING: This module has been deprecated.
+
+Reason:
+This module is no longer recommended for use, as it is replaced by the function softwareVersionsToYAML
+in the utils_nfcore_pipeline subworkflow that is included in the nf-core template.
+
+"""
 process CUSTOM_DUMPSOFTWAREVERSIONS {
     label 'process_single'
 
     // Requires `pyyaml` which does not have a dedicated container but is in the MultiQC container
     conda "${moduleDir}/environment.yml"
     container "${ workflow.containerEngine == 'singularity' && !task.ext.singularity_pull_docker_container ?
-        'https://depot.galaxyproject.org/singularity/multiqc:1.20--pyhdfd78af_0' :
-        'biocontainers/multiqc:1.20--pyhdfd78af_0' }"
+        'https://depot.galaxyproject.org/singularity/multiqc:1.27--pyhdfd78af_0' :
+        'biocontainers/multiqc:1.27--pyhdfd78af_0' }"
 
     input:
     path versions
@@ -19,6 +27,7 @@ process CUSTOM_DUMPSOFTWAREVERSIONS {
     task.ext.when == null || task.ext.when
 
     script:
+    assert true: deprecation_message
     def args = task.ext.args ?: ''
     template 'dumpsoftwareversions.py'
 }
diff --git a/modules/nf-core/custom/dumpsoftwareversions/tests/main.nf.test.snap b/modules/nf-core/custom/dumpsoftwareversions/tests/main.nf.test.snap
index 5f59a936..74e42fb1 100644
--- a/modules/nf-core/custom/dumpsoftwareversions/tests/main.nf.test.snap
+++ b/modules/nf-core/custom/dumpsoftwareversions/tests/main.nf.test.snap
@@ -2,7 +2,7 @@
     "Should run without failures": {
         "content": [
             [
-                "versions.yml:md5,76d454d92244589d32455833f7c1ba6d"
+                "versions.yml:md5,42bedca466554ea4ad0b586f8a18be28"
             ],
             [
                 "data: \"<style>\\n#nf-core-versions tbody:nth-child(even) {\\n    background-color: #f2f2f2;\\n\\",
@@ -10,8 +10,8 @@
                 "  >\\n    <thead>\\n        <tr>\\n            <th> Process Name </th>\\n            <th>\\",
                 "  \\ Software </th>\\n            <th> Version  </th>\\n        </tr>\\n    </thead>\\n\\",
                 "  \\n<tbody>\\n<tr>\\n    <td><samp>CUSTOM_DUMPSOFTWAREVERSIONS</samp></td>\\n    <td><samp>python</samp></td>\\n\\",
-                "  \\    <td><samp>3.11.7</samp></td>\\n</tr>\\n\\n<tr>\\n    <td><samp></samp></td>\\n \\",
-                "  \\   <td><samp>yaml</samp></td>\\n    <td><samp>5.4.1</samp></td>\\n</tr>\\n\\n</tbody>\\n\\",
+                "  \\    <td><samp>3.13.1</samp></td>\\n</tr>\\n\\n<tr>\\n    <td><samp></samp></td>\\n \\",
+                "  \\   <td><samp>yaml</samp></td>\\n    <td><samp>6.0.2</samp></td>\\n</tr>\\n\\n</tbody>\\n\\",
                 "  <tbody>\\n<tr>\\n    <td><samp>TOOL1</samp></td>\\n    <td><samp>tool1</samp></td>\\n\\",
                 "  \\    <td><samp>0.11.9</samp></td>\\n</tr>\\n\\n</tbody>\\n<tbody>\\n<tr>\\n    <td><samp>TOOL2</samp></td>\\n\\",
                 "  \\    <td><samp>tool2</samp></td>\\n    <td><samp>1.9</samp></td>\\n</tr>\\n\\n</tbody>\\n\\",
@@ -19,8 +19,8 @@
             ],
             [
                 "CUSTOM_DUMPSOFTWAREVERSIONS:",
-                "  python: 3.11.7",
-                "  yaml: 5.4.1",
+                "  python: 3.13.1",
+                "  yaml: 6.0.2",
                 "TOOL1:",
                 "  tool1: 0.11.9",
                 "TOOL2:",
@@ -28,6 +28,10 @@
                 "Workflow:"
             ]
         ],
-        "timestamp": "2024-01-09T23:01:18.710682"
+        "meta": {
+            "nf-test": "0.9.2",
+            "nextflow": "24.10.5"
+        },
+        "timestamp": "2025-03-11T13:50:29.789124124"
     }
 }
\ No newline at end of file
diff --git a/modules/nf-core/custom/dumpsoftwareversions/tests/tags.yml b/modules/nf-core/custom/dumpsoftwareversions/tests/tags.yml
deleted file mode 100644
index 405aa24a..00000000
--- a/modules/nf-core/custom/dumpsoftwareversions/tests/tags.yml
+++ /dev/null
@@ -1,2 +0,0 @@
-custom/dumpsoftwareversions:
-  - modules/nf-core/custom/dumpsoftwareversions/**
diff --git a/subworkflows/janelia/spark_start/main.nf b/subworkflows/janelia/spark_start/main.nf
index 56f08265..59643c95 100644
--- a/subworkflows/janelia/spark_start/main.nf
+++ b/subworkflows/janelia/spark_start/main.nf
@@ -1,6 +1,42 @@
+process PREPARE_SPARK_CONFIG {
+    label 'process_single'
+    container 'ghcr.io/janeliascicomp/spark:3.3.2-scala2.12-java17-ubuntu24.04'
+
+    input:
+    tuple val(meta), val(spark), path(spark_work_dir)
+    val(spark_config)
+
+    output:
+    tuple val(meta), val(spark), env(full_spark_work_dir), emit: spark_inputs
+    env(spark_config_filepath)                           , emit: spark_config_file
+
+    script:
+    def spark_local_dir = task.ext.spark_local_dir
+        ? file(task.ext.spark_local_dir)
+        : "/tmp/spark-${workflow.sessionId}"
+
+    def spark_config_content = spark_config.inject('# Spark configuration') { acc, k, v ->
+        "${acc}\n${k}=${v}"
+    }
+    """
+    full_spark_local_dir=\$(readlink -m ${spark_local_dir})
+    full_spark_work_dir=\$(readlink -m ${spark_work_dir})
+    if [[ ! -e \${full_spark_work_dir} ]] ; then
+        echo "Create spark work directory ${spark_work_dir} -> \${full_spark_work_dir}"
+        mkdir -p \${full_spark_work_dir}
+    else
+        echo "Spark work directory: \${full_spark_work_dir} - already exists"
+    fi
+    spark_config_filepath="\${full_spark_work_dir}/spark-defaults.conf"
+    echo "Create spark config file \${spark_config_filepath}"
+    echo "${spark_config_content}" > \${spark_config_filepath}
+    echo "spark.local.dir=\${full_spark_local_dir}" >> \${spark_config_filepath}
+    """
+}
+
 process SPARK_STARTMANAGER {
     label 'process_long'
-    container 'ghcr.io/janeliascicomp/spark:3.1.3'
+    container 'ghcr.io/janeliascicomp/spark:3.3.2-scala2.12-java17-ubuntu24.04'
 
     input:
     tuple val(meta), val(spark), path(spark_work_dir)
@@ -12,14 +48,17 @@ process SPARK_STARTMANAGER {
     task.ext.when == null || task.ext.when
 
     script:
-    args = task.ext.args ?: ''
-    spark_local_dir = task.ext.spark_local_dir ?: "/tmp/spark-${workflow.sessionId}"
-    sleep_secs = task.ext.sleep_secs ?: '1'
-    spark_config_filepath = "\${full_spark_work_dir}/spark-defaults.conf"
-    spark_master_log_file = "\${full_spark_work_dir}/sparkmaster.log"
-    terminate_file_name = "\${full_spark_work_dir}/terminate-spark"
-    container_engine = workflow.containerEngine
+    def args = task.ext.args ?: ''
+    def spark_local_dir = task.ext.spark_local_dir
+        ? file(task.ext.spark_local_dir)
+        : "/tmp/spark-${workflow.sessionId}"
+    def sleep_secs = task.ext.sleep_secs ?: '1'
+    def spark_config_filepath = "\${full_spark_work_dir}/spark-defaults.conf"
+    def spark_master_log_file = "\${full_spark_work_dir}/sparkmaster.log"
+    def terminate_file_name = "\${full_spark_work_dir}/terminate-spark"
+    def container_engine = workflow.containerEngine
     """
+    full_spark_local_dir=\$(readlink -m ${spark_local_dir})
     full_spark_work_dir=\$(readlink -m ${spark_work_dir})
     if [[ ! -e \${full_spark_work_dir} ]] ; then
         echo "Create spark work directory ${spark_work_dir} -> \${full_spark_work_dir}"
@@ -28,17 +67,29 @@ process SPARK_STARTMANAGER {
         echo "Spark work directory: \${full_spark_work_dir} - already exists"
     fi
 
-    /opt/scripts/startmanager.sh "$spark_local_dir" \${full_spark_work_dir} "$spark_master_log_file" \
-        "$spark_config_filepath" "$terminate_file_name" "$args" $sleep_secs $container_engine
+    CMD=(
+        /opt/scripts/startmanager.sh
+        "\${full_spark_local_dir}"
+        "\${full_spark_work_dir}"
+        "$spark_master_log_file"
+        "$spark_config_filepath"
+        "$terminate_file_name"
+        "$args"
+        $sleep_secs
+        $container_engine
+    )
+    echo "CMD: \${CMD[@]}"
+    (exec "\${CMD[@]}")
     """
 }
 
 process SPARK_WAITFORMANAGER {
     label 'process_single'
-    container 'ghcr.io/janeliascicomp/spark:3.1.3'
+    container 'ghcr.io/janeliascicomp/spark:3.3.2-scala2.12-java17-ubuntu24.04'
     errorStrategy { task.exitStatus == 2
         ? 'retry' // retry on a timeout to prevent the case when the waiter is started before the master and master never gets its chance
-        : 'terminate' }
+        : 'terminate'
+    }
     maxRetries 20
 
     input:
@@ -51,21 +102,33 @@ process SPARK_WAITFORMANAGER {
     task.ext.when == null || task.ext.when
 
     script:
-    sleep_secs = task.ext.sleep_secs ?: '1'
-    max_wait_secs = task.ext.max_wait_secs ?: '3600'
-    spark_master_log_name = "\${full_spark_work_dir}/sparkmaster.log"
-    terminate_file_name = "\${full_spark_work_dir}/terminate-spark"
+    def sleep_secs = task.ext.sleep_secs ?: '1'
+    def max_wait_secs = task.ext.max_wait_secs ?: '3600'
+    def spark_master_log_name = "\${full_spark_work_dir}/sparkmaster.log"
+    def terminate_file_name = "\${full_spark_work_dir}/terminate-spark"
     """
     full_spark_work_dir=\$(readlink -m ${spark_work_dir})
 
-    /opt/scripts/waitformanager.sh "$spark_master_log_name" "$terminate_file_name" $sleep_secs $max_wait_secs
-    export spark_uri=`cat spark_uri`
+    CMD=(
+        /opt/scripts/waitformanager.sh
+        "$spark_master_log_name"
+        "$terminate_file_name"
+        $sleep_secs
+        $max_wait_secs
+    )
+
+    echo "CMD: \${CMD[@]}"
+    (exec "\${CMD[@]}")
+
+    spark_uri=\$(cat spark_uri)
+
+    echo "Export spark URI: \${spark_uri}"
     """
 }
 
 process SPARK_STARTWORKER {
     label 'process_long'
-    container 'ghcr.io/janeliascicomp/spark:3.1.3'
+    container 'ghcr.io/janeliascicomp/spark:3.3.2-scala2.12-java17-ubuntu24.04'
     cpus { spark.worker_cores }
     memory { spark.worker_memory }
 
@@ -80,26 +143,38 @@ process SPARK_STARTWORKER {
     task.ext.when == null || task.ext.when
 
     script:
-    args = task.ext.args ?: ''
-    sleep_secs = task.ext.sleep_secs ?: '1'
-    spark_worker_log_file = "\${full_spark_work_dir}/sparkworker-${worker_id}.log"
-    spark_config_filepath = "\${full_spark_work_dir}/spark-defaults.conf"
-    terminate_file_name = "\${full_spark_work_dir}/terminate-spark"
-    worker_memory = spark.worker_memory.replace(" KB",'').replace(" MB",'').replace(" GB",'').replace(" TB",'')
-    container_engine = workflow.containerEngine
+    def args = task.ext.args ?: ''
+    def sleep_secs = task.ext.sleep_secs ?: '1'
+    def spark_worker_log_file = "\${full_spark_work_dir}/sparkworker-${worker_id}.log"
+    def spark_config_filepath = "\${full_spark_work_dir}/spark-defaults.conf"
+    def terminate_file_name = "\${full_spark_work_dir}/terminate-spark"
+    def worker_memory = spark.worker_memory.replace(" KB",'').replace(" MB",'').replace(" GB",'').replace(" TB",'')
+    def container_engine = workflow.containerEngine
     """
     full_spark_work_dir=\$(readlink -m ${spark_work_dir})
 
-    /opt/scripts/startworker.sh "\${full_spark_work_dir}" "${spark.uri}" $worker_id \
-        ${spark.worker_cores} ${worker_memory} \
-        "$spark_worker_log_file" "$spark_config_filepath" "$terminate_file_name" \
-        "$args" $sleep_secs $container_engine
+    CMD=(
+        /opt/scripts/startworker.sh
+        "\${full_spark_work_dir}"
+        "${spark.uri}"
+        "$worker_id"
+        "${spark.worker_cores}"
+        "${worker_memory}"
+        "$spark_worker_log_file"
+        "$spark_config_filepath"
+        "$terminate_file_name"
+        "$args"
+        "$sleep_secs"
+        "$container_engine"
+    )
+    echo "CMD: \${CMD[@]}"
+    (exec "\${CMD[@]}")
     """
 }
 
 process SPARK_WAITFORWORKER {
     label 'process_single'
-    container 'ghcr.io/janeliascicomp/spark:3.1.3'
+    container 'ghcr.io/janeliascicomp/spark:3.3.2-scala2.12-java17-ubuntu24.04'
     // retry on a timeout to prevent the case when the waiter is started
     // before the worker and the worker never gets its chance
     errorStrategy { task.exitStatus == 2 ? 'retry' : 'terminate' }
@@ -115,22 +190,28 @@ process SPARK_WAITFORWORKER {
     task.ext.when == null || task.ext.when
 
     script:
-    sleep_secs = task.ext.sleep_secs ?: '1'
-    max_wait_secs = task.ext.max_wait_secs ?: '3600'
-    spark_worker_log_file = "\${full_spark_work_dir}/sparkworker-${worker_id}.log"
-    terminate_file_name = "\${full_spark_work_dir}/terminate-spark"
+    def sleep_secs = task.ext.sleep_secs ?: '1'
+    def max_wait_secs = task.ext.max_wait_secs ?: '3600'
+    def spark_worker_log_file = "\${full_spark_work_dir}/sparkworker-${worker_id}.log"
+    def terminate_file_name = "\${full_spark_work_dir}/terminate-spark"
     """
     full_spark_work_dir=\$(readlink -m ${spark_work_dir})
 
-    /opt/scripts/waitforworker.sh "${spark.uri}" \
-        "$spark_worker_log_file" "$terminate_file_name" \
-        $sleep_secs $max_wait_secs
+    CMD=(
+        /opt/scripts/waitforworker.sh
+        "${spark.uri}"
+        "$spark_worker_log_file"
+        "$terminate_file_name"
+        $sleep_secs
+        $max_wait_secs
+    )
+    (exec "\${CMD[@]}")
     """
 }
 
 process SPARK_CLEANUP {
     label 'process_single'
-    container 'ghcr.io/janeliascicomp/spark:3.1.3'
+    container 'ghcr.io/janeliascicomp/spark:3.3.2-scala2.12-java17-ubuntu24.04'
 
     input:
     tuple val(meta), val(spark), path(spark_work_dir)
@@ -153,6 +234,7 @@ process SPARK_CLEANUP {
 workflow SPARK_START {
     take:
     ch_meta               // channel: [ meta, [data_paths] ]
+    config                // map: spark configuration options
     spark_cluster         // boolean: use a distributed cluster?
     working_dir           // path: shared storage path for worker communication
     spark_workers         // int: number of workers in the cluster (ignored if spark_cluster is false)
@@ -165,21 +247,31 @@ workflow SPARK_START {
     main:
     // create a Spark context for each meta
     def n_spark_workers = spark_workers ?: 1
-    def meta_and_spark = ch_meta
+
+    def spark_config = config + [
+        'spark.rpc.askTimeout': '300s',
+        'spark.storage.blockManagerHeartBeatMs': '30000',
+        'spark.rpc.retry.wait': '30s',
+        'spark.kryoserializer.buffer.max': '1024m',
+        'spark.core.connection.ack.wait.timeout': '600s',
+        'spark.driver.maxResultSize': '0',
+        'spark.worker.cleanup.enabled': 'true',
+    ]
+
+    def spark_inputs = ch_meta
     | map {
-        def (meta, data_paths) = it // ch_meta
+        def (meta) = it // ch_meta: [ meta, [data_paths] ] - ignore data_paths
         def spark_work_dir = file("${working_dir}/spark/${meta.id}")
-        def spark = [:]
-        spark.work_dir = spark_work_dir
-        spark.workers = n_spark_workers
-        spark.worker_cores = spark_worker_cpus ?: 1
-        spark.driver_cores = spark_driver_cpus ?: 1
-        def spark_driver_mem_gb_value = spark_driver_mem_gb ?: (spark.driver_cores * spark_gb_per_cpu)
-        spark.driver_memory = "${spark_driver_mem_gb_value} GB"
-        spark.parallelism = (n_spark_workers * spark_worker_cpus)
-        // 1 GB of overhead for Spark, the rest for executors
-        spark.worker_memory = (spark_worker_cpus * spark_gb_per_cpu + 1)+" GB"
-        spark.executor_memory = (spark_worker_cpus * spark_gb_per_cpu)+" GB"
+        def spark = [
+            work_dir: spark_work_dir,
+            workers: n_spark_workers,
+            worker_cores: spark_worker_cpus ?: 1,
+            driver_cores: spark_driver_cpus ?: 1,
+            driver_memory: "${spark_driver_mem_gb ?: (spark.driver_cores * spark_gb_per_cpu)} GB",
+            parallelism: (n_spark_workers * spark_worker_cpus),
+            worker_memory: (spark_worker_cpus * spark_gb_per_cpu + 1)+" GB", // 1 GB of overhead for Spark, the rest for executors
+            executor_memory: (spark_worker_cpus * spark_gb_per_cpu)+" GB",
+        ]
         def r = [
             meta,
             spark,
@@ -189,13 +281,15 @@ workflow SPARK_START {
         r
     }
 
+    PREPARE_SPARK_CONFIG(spark_inputs, spark_config)
+
     if (spark_cluster) {
         // start the Spark manager
         // this runs indefinitely until SPARK_TERMINATE is called
-        SPARK_STARTMANAGER(meta_and_spark)
+        SPARK_STARTMANAGER(PREPARE_SPARK_CONFIG.out.spark_inputs)
 
         // start a watcher that waits for the manager to be ready
-        SPARK_WAITFORMANAGER(meta_and_spark)
+        SPARK_WAITFORMANAGER(PREPARE_SPARK_CONFIG.out.spark_inputs)
 
         // prepare all arguments for all workers
         def meta_workers_with_data = SPARK_WAITFORMANAGER.out
@@ -218,7 +312,10 @@ workflow SPARK_START {
 
         // start workers
         // these run indefinitely until SPARK_TERMINATE is called
-        SPARK_STARTWORKER(meta_workers_with_data.worker, meta_workers_with_data.data)
+        SPARK_STARTWORKER(
+            meta_workers_with_data.worker,
+            meta_workers_with_data.data,
+        )
         SPARK_STARTWORKER.out.groupTuple(by:[0,1,2], size: n_spark_workers)
         | map {
             def (meta, spark, spark_work_dir, worker_ids) = it
@@ -240,7 +337,7 @@ workflow SPARK_START {
         }
     } else {
         // when running locally, the driver needs enough resources to run a spark worker
-        spark_context = meta_and_spark.map {
+        spark_context = PREPARE_SPARK_CONFIG.out.spark_inputs.map {
             def (meta, spark, spark_work_dir) = it
             spark.workers = 1
             spark.driver_cores = spark_driver_cpus + spark_worker_cpus
diff --git a/subworkflows/local/multiscale.nf b/subworkflows/local/multiscale.nf
index adf8559d..f31d591f 100644
--- a/subworkflows/local/multiscale.nf
+++ b/subworkflows/local/multiscale.nf
@@ -33,7 +33,8 @@ workflow MULTISCALE {
 
     if (!skip) {
         def downsample_input = SPARK_START(
-            prepared_data, // [meta, data_paths]
+            prepared_data,     // [meta, data_paths]
+            [:],               // spark default config
             with_spark_cluster,
             workdir,
             spark_workers,
diff --git a/subworkflows/local/stitching.nf b/subworkflows/local/stitching.nf
index 45e473f8..48a4b04f 100644
--- a/subworkflows/local/stitching.nf
+++ b/subworkflows/local/stitching.nf
@@ -62,7 +62,8 @@ workflow STITCHING {
 
     if (!skip) {
         def stitching_input = SPARK_START(
-            prepared_data, // [meta, data_paths]
+            prepared_data,      // [meta, data_paths]
+            [:],                // spark default config
             with_spark_cluster,
             workdir,
             spark_workers,
diff --git a/workflows/registration.nf b/workflows/registration.nf
index a367d83c..2d8f6c65 100644
--- a/workflows/registration.nf
+++ b/workflows/registration.nf
@@ -768,6 +768,7 @@ workflow RUN_MULTISCALE_WITH_SINGLE_CLUSTER {
     if (!params.skip_multiscale) {
         def downsample_input = SPARK_START(
             multiscale_cluster_data,
+            [:],                                                               // spark default config
             params.multiscale_with_spark_cluster,
             multiscale_work_dir,
             params.multiscale_spark_workers ?: params.spark_workers,
diff --git a/workflows/spot_extraction.nf b/workflows/spot_extraction.nf
index 3d22baad..59c1fc4e 100644
--- a/workflows/spot_extraction.nf
+++ b/workflows/spot_extraction.nf
@@ -58,6 +58,7 @@ workflow SPOT_EXTRACTION {
 
         def rsfish_input = SPARK_START(
             spots_spark_input,
+            [:],                                // spark default config
             params.distributed_spot_extraction,
             workdir,
             params.rsfish_spark_workers,

From 412875015350b3e1461bac156c705a8f403459e4 Mon Sep 17 00:00:00 2001
From: Cristian Goina <goinac@hhmi.org>
Date: Tue, 15 Apr 2025 18:01:48 -0400
Subject: [PATCH 2/8] set container to be used for spark stitching tasks

---
 conf/stitching-modules.config | 12 ++++++++----
 1 file changed, 8 insertions(+), 4 deletions(-)

diff --git a/conf/stitching-modules.config b/conf/stitching-modules.config
index bca7dc8b..dee03048 100644
--- a/conf/stitching-modules.config
+++ b/conf/stitching-modules.config
@@ -33,25 +33,29 @@ process {
         errorStrategy = 'terminate'
     }
 
-    withName: "(.*)?SPARK_STARTMANAGER" {
+    withName: "(.*)?STITCHING:(.*):SPARK_STARTMANAGER" {
         errorStrategy = 'terminate'
+        ext.container = 'ghcr.io/janeliascicomp/stitching-spark:1.11.0-rc2'
         ext.spark_local_dir = params.spark_local_dir ? params.spark_local_dir : ''
         ext.sleep_secs = 2
     }
 
-    withName: "(.*)?SPARK_STARTWORKER" {
+    withName: "(.*)?STITCHING:(.*):SPARK_STARTWORKER" {
         errorStrategy = 'terminate'
+        ext.container = 'ghcr.io/janeliascicomp/stitching-spark:1.11.0-rc2'
         ext.sleep_secs = 2
     }
 
-    withName: "(.*)?SPARK_WAITFORMANAGER" {
+    withName: "(.*)?STITCHING:(.*):SPARK_WAITFORMANAGER" {
         errorStrategy = 'terminate'
+        ext.container = 'ghcr.io/janeliascicomp/stitching-spark:1.11.0-rc2'
         ext.sleep_secs = 2
         ext.max_wait_secs = 7200
     }
 
-    withName: "(.*)?SPARK_WAITFORWORKER" {
+    withName: "(.*)?STITCHING:(.*):SPARK_WAITFORWORKER" {
         errorStrategy = 'terminate'
+        ext.container = 'ghcr.io/janeliascicomp/stitching-spark:1.11.0-rc2'
         ext.sleep_secs = 2
         ext.max_wait_secs = 7200
     }

From 83952e59bec954593ce99aaba76ad7a3b0effa08 Mon Sep 17 00:00:00 2001
From: Cristian Goina <goinac@hhmi.org>
Date: Tue, 15 Apr 2025 18:10:47 -0400
Subject: [PATCH 3/8] set container not ext.container

---
 conf/stitching-modules.config | 8 ++------
 1 file changed, 2 insertions(+), 6 deletions(-)

diff --git a/conf/stitching-modules.config b/conf/stitching-modules.config
index dee03048..b80f38d7 100644
--- a/conf/stitching-modules.config
+++ b/conf/stitching-modules.config
@@ -29,33 +29,29 @@ process {
     withName:"(.*)?STITCHING:.*" {
         // all processes that run as part of the stitching workflow
         // use this container
-        ext.container = 'ghcr.io/janeliascicomp/stitching-spark:1.11.0-rc2'
+        container = 'ghcr.io/janeliascicomp/stitching-spark:1.11.0-rc2'
         errorStrategy = 'terminate'
     }
 
-    withName: "(.*)?STITCHING:(.*):SPARK_STARTMANAGER" {
+    withName: "(.*)?STITCHING:(.*)SPARK_STARTMANAGER" {
         errorStrategy = 'terminate'
-        ext.container = 'ghcr.io/janeliascicomp/stitching-spark:1.11.0-rc2'
         ext.spark_local_dir = params.spark_local_dir ? params.spark_local_dir : ''
         ext.sleep_secs = 2
     }
 
     withName: "(.*)?STITCHING:(.*):SPARK_STARTWORKER" {
         errorStrategy = 'terminate'
-        ext.container = 'ghcr.io/janeliascicomp/stitching-spark:1.11.0-rc2'
         ext.sleep_secs = 2
     }
 
     withName: "(.*)?STITCHING:(.*):SPARK_WAITFORMANAGER" {
         errorStrategy = 'terminate'
-        ext.container = 'ghcr.io/janeliascicomp/stitching-spark:1.11.0-rc2'
         ext.sleep_secs = 2
         ext.max_wait_secs = 7200
     }
 
     withName: "(.*)?STITCHING:(.*):SPARK_WAITFORWORKER" {
         errorStrategy = 'terminate'
-        ext.container = 'ghcr.io/janeliascicomp/stitching-spark:1.11.0-rc2'
         ext.sleep_secs = 2
         ext.max_wait_secs = 7200
     }

From d58c66ad8ad2ff00702dd247e23ef4fe842332f0 Mon Sep 17 00:00:00 2001
From: Cristian Goina <goinac@hhmi.org>
Date: Tue, 15 Apr 2025 18:28:00 -0400
Subject: [PATCH 4/8] set container for spot extraction tasks

---
 conf/spot-extraction-modules.config | 7 +++++++
 modules/janelia/rs_fish/main.nf     | 2 +-
 2 files changed, 8 insertions(+), 1 deletion(-)

diff --git a/conf/spot-extraction-modules.config b/conf/spot-extraction-modules.config
index 1ee169a5..14b29f35 100644
--- a/conf/spot-extraction-modules.config
+++ b/conf/spot-extraction-modules.config
@@ -28,6 +28,13 @@ params {
 
 process {
 
+    withName:"(.*)?SPOT_EXTRACTION:.*" {
+        // all processes that run as part of the stitching workflow
+        // use this container
+        container = 'ghcr.io/janeliascicomp/rs-fish-spark:4d1b901'
+        errorStrategy = 'terminate'
+    }
+
     withName: "(.*)?RS_FISH" {
         errorStrategy = 'terminate'
         ext.container = 'ghcr.io/janeliascicomp/rs-fish-spark:4d1b901'
diff --git a/modules/janelia/rs_fish/main.nf b/modules/janelia/rs_fish/main.nf
index 2028f7a0..c5a96297 100755
--- a/modules/janelia/rs_fish/main.nf
+++ b/modules/janelia/rs_fish/main.nf
@@ -1,6 +1,6 @@
 process RS_FISH {
     tag "${meta.id}"
-    container { task && task.ext.container ?: 'ghcr.io/janeliascicomp/rs-fish-spark:8f8954f' }
+    container { task && task.ext.container ?: 'ghcr.io/janeliascicomp/rs-fish-spark:4d1b901' }
     cpus { spark.driver_cores }
     memory { spark.driver_memory }
 

From 126542d00ba83ca65effb7ae339a1be1c12292b0 Mon Sep 17 00:00:00 2001
From: Cristian Goina <goinac@hhmi.org>
Date: Tue, 15 Apr 2025 18:37:00 -0400
Subject: [PATCH 5/8] set different container for post_rs_fish

---
 conf/spot-extraction-modules.config | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/conf/spot-extraction-modules.config b/conf/spot-extraction-modules.config
index 14b29f35..76eeb16c 100644
--- a/conf/spot-extraction-modules.config
+++ b/conf/spot-extraction-modules.config
@@ -35,7 +35,7 @@ process {
         errorStrategy = 'terminate'
     }
 
-    withName: "(.*)?RS_FISH" {
+    withName: "(.*)?SPOT_EXTRACTION:RS_FISH" {
         errorStrategy = 'terminate'
         ext.container = 'ghcr.io/janeliascicomp/rs-fish-spark:4d1b901'
         ext.args = [
@@ -48,6 +48,10 @@ process {
         ].join(' ')
     }
 
+    withName: "(.*)?SPOT_EXTRACTION:POST_RS_FISH" {
+        ext.container = 'ghcr.io/janeliascicomp/post-rs-fish:v1'
+    }
+
 }
 
 def arg_value(arg_flag, arg_value) {

From 036aa41530fbba94792f7472b7123fe0c609eb1f Mon Sep 17 00:00:00 2001
From: Cristian Goina <goinac@hhmi.org>
Date: Tue, 15 Apr 2025 18:40:01 -0400
Subject: [PATCH 6/8] changed regex for spot_extraction spark tasks

---
 conf/spot-extraction-modules.config | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/conf/spot-extraction-modules.config b/conf/spot-extraction-modules.config
index 76eeb16c..c2978365 100644
--- a/conf/spot-extraction-modules.config
+++ b/conf/spot-extraction-modules.config
@@ -28,9 +28,8 @@ params {
 
 process {
 
-    withName:"(.*)?SPOT_EXTRACTION:.*" {
-        // all processes that run as part of the stitching workflow
-        // use this container
+    withName:"(.*)?SPOT_EXTRACTION:SPARK.*" {
+        // set container for all spark tasks
         container = 'ghcr.io/janeliascicomp/rs-fish-spark:4d1b901'
         errorStrategy = 'terminate'
     }

From 2e2588605d4345d2044d84e160d68ea2abaf9de7 Mon Sep 17 00:00:00 2001
From: Cristian Goina <goinac@janelia.hhmi.org>
Date: Tue, 15 Apr 2025 18:52:02 -0400
Subject: [PATCH 7/8] set rs-fish container

---
 conf/spot-extraction-modules.config | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/conf/spot-extraction-modules.config b/conf/spot-extraction-modules.config
index c2978365..ae160ed1 100644
--- a/conf/spot-extraction-modules.config
+++ b/conf/spot-extraction-modules.config
@@ -28,7 +28,7 @@ params {
 
 process {
 
-    withName:"(.*)?SPOT_EXTRACTION:SPARK.*" {
+    withName: "(.*)?SPOT_EXTRACTION:SPARK.*" {
         // set container for all spark tasks
         container = 'ghcr.io/janeliascicomp/rs-fish-spark:4d1b901'
         errorStrategy = 'terminate'
@@ -36,7 +36,7 @@ process {
 
     withName: "(.*)?SPOT_EXTRACTION:RS_FISH" {
         errorStrategy = 'terminate'
-        ext.container = 'ghcr.io/janeliascicomp/rs-fish-spark:4d1b901'
+        container = 'ghcr.io/janeliascicomp/rs-fish-spark:4d1b901'
         ext.args = [
             "--minIntensity=${params.rsfish_min_intensity}",
             "--maxIntensity=${params.rsfish_max_intensity}",
@@ -48,7 +48,7 @@ process {
     }
 
     withName: "(.*)?SPOT_EXTRACTION:POST_RS_FISH" {
-        ext.container = 'ghcr.io/janeliascicomp/post-rs-fish:v1'
+        container = 'ghcr.io/janeliascicomp/post-rs-fish:v1'
     }
 
 }

From 6f562fc2c9d1656e2172decbb7531d92faf8f2d8 Mon Sep 17 00:00:00 2001
From: Cristian Goina <goinac@hhmi.org>
Date: Tue, 15 Apr 2025 18:57:13 -0400
Subject: [PATCH 8/8] fixed driver mem

---
 conf/stitching-modules.config            | 2 +-
 modules.json                             | 2 +-
 subworkflows/janelia/spark_start/main.nf | 2 +-
 3 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/conf/stitching-modules.config b/conf/stitching-modules.config
index b80f38d7..842af5c6 100644
--- a/conf/stitching-modules.config
+++ b/conf/stitching-modules.config
@@ -26,7 +26,7 @@ params {
 
 process {
 
-    withName:"(.*)?STITCHING:.*" {
+    withName: "(.*)?STITCHING:.*" {
         // all processes that run as part of the stitching workflow
         // use this container
         container = 'ghcr.io/janeliascicomp/stitching-spark:1.11.0-rc2'
diff --git a/modules.json b/modules.json
index d7cc8937..8392d1f0 100644
--- a/modules.json
+++ b/modules.json
@@ -66,7 +66,7 @@
                     },
                     "spark_start": {
                         "branch": "main",
-                        "git_sha": "cfc0afa06f34e7c94d875bdbafb45889d9405cd2",
+                        "git_sha": "e3b9a6883f427176c180f32e45b268539bd5496c",
                         "installed_by": ["subworkflows"]
                     },
                     "spark_stop": {
diff --git a/subworkflows/janelia/spark_start/main.nf b/subworkflows/janelia/spark_start/main.nf
index 59643c95..89431a3c 100644
--- a/subworkflows/janelia/spark_start/main.nf
+++ b/subworkflows/janelia/spark_start/main.nf
@@ -267,7 +267,7 @@ workflow SPARK_START {
             workers: n_spark_workers,
             worker_cores: spark_worker_cpus ?: 1,
             driver_cores: spark_driver_cpus ?: 1,
-            driver_memory: "${spark_driver_mem_gb ?: (spark.driver_cores * spark_gb_per_cpu)} GB",
+            driver_memory: "${spark_driver_mem_gb ?: (spark_driver_cpus * spark_gb_per_cpu)} GB",
             parallelism: (n_spark_workers * spark_worker_cpus),
             worker_memory: (spark_worker_cpus * spark_gb_per_cpu + 1)+" GB", // 1 GB of overhead for Spark, the rest for executors
             executor_memory: (spark_worker_cpus * spark_gb_per_cpu)+" GB",