Executor for Apache Spark

[rbavery]

Could Spark be added as a supported executor?

Maybe RDD.map or RDD.mapPartitions would be the correct way to map a function similar to map_unordered in the Lithops executor.

https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.RDD.mapPartitions.html#pyspark.RDD.mapPartitions

To support this a guess would need to be made up front on the reserved memory available for python UDFs. It sounds like currently this would be done globally but maybe later could be done on a per-operator basis?

[tomwhite]

A Spark executor would be a great addition. I just added some notes about implementing a new executor in #498 if you're interested in having a go at this @rbavery?

[rbavery]

I'm definitely interested, thanks for adding notes. It's possible I won't make quick (or any) progress because of other responsibilities 😬

[songhan89]

Had a quick stab with Tom's notes and with some help from DeepSeeker R1 .

Will try to experiment and test further.

Work in progress

from pyspark import SparkContext, SparkConf
from cubed.runtime.types import DagExecutor, TaskEndEvent
from cubed.runtime.pipeline import visit_nodes
from cubed.runtime.utils import handle_operation_start_callbacks

class SparkExecutor(DagExecutor):
    """Execute Cubed DAG using Apache Spark (supports reductions)."""
    @property
    def name(self) -> str:
        return "spark"

    def execute_dag(
        self,
        dag,
        callbacks=None,
        resume=None,
        spec=None,
        compute_id=None,
        **kwargs,
    ):
        conf = SparkConf().setAppName("CubedSparkExecutor")
        if spec is not None:
            executor_mem = (spec.allowed_mem + spec.reserved_mem) // (1024 ** 2)
            conf.set("spark.executor.memory", f"{executor_mem}m")
            conf.set("spark.speculation", "true")

        sc = SparkContext.getOrCreate(conf=conf)

        for name, node in visit_nodes(dag, resume=resume):
            handle_operation_start_callbacks(callbacks, name)
            pipeline = node["pipeline"]
            mappable = list(pipeline.mappable)

            # Process tasks and collect results
            rdd = sc.parallelize(mappable, numSlices=len(mappable))
            results = rdd.map(
                lambda x: pipeline.function(x, config=pipeline.config)
            ).collect()  # <-- Collect results for reductions

            # Fire callbacks for all tasks in this node
            if callbacks is not None:
                event = TaskEndEvent(name=name, num_tasks=len(results))
                [callback.on_task_end(event) for callback in callbacks]

            # Propagate results to subsequent nodes
            # (Cubed handles this internally; this is a placeholder)
            node["result"] = results

Results running on Databricks Spark Executor

[songhan89]

Hi @tomwhite ,

Would like to get some pointers/advice on the implementation of lazy loading and compute for Spark. Right now I only know how to implement eager loading lazy_rdd.collect(). Any idea how best i should approach this ?

from typing import Any, Optional, Sequence, List, Tuple, Dict, Union
from networkx import MultiDiGraph
from pyspark.sql import SparkSession

from cubed.runtime.pipeline import visit_nodes 
from cubed.runtime.types import Callback, DagExecutor 
from cubed.runtime.utils import handle_operation_start_callbacks, handle_callbacks  
from cubed.spec import Spec 


class SparkExecutor(DagExecutor):
    """An execution engine that uses Apache Spark."""

    @property
    def name(self) -> str:
        return "spark"

    def execute_dag(
        self,
        dag: MultiDiGraph,
        callbacks: Optional[Sequence[Callback]] = None,
        resume: Optional[bool] = None,
        spec: Optional[Spec] = None,
        compute_id: Optional[str] = None,
        **kwargs: Any,
    ):

        # Configure Spark memory settings from Spec if provided
        spark_builder = SparkSession.builder
        if spec is not None and hasattr(spec, "allowed_mem") and spec.allowed_mem:
            spark_builder = spark_builder.config("spark.executor.memory", spec.allowed_mem)
            spark_builder = spark_builder.config("spark.driver.memory", spec.allowed_mem)
            spark_builder = spark_builder.config("spark.speculation", "true")
            
        # Create a Spark session
        spark = spark_builder.getOrCreate()

        for name, node in visit_nodes(dag, resume=resume):
            handle_operation_start_callbacks(callbacks, name)
            pipeline = node["pipeline"]
            # Create an RDD from pipeline.mappable.
            rdd = spark.sparkContext.parallelize(pipeline.mappable)
            # Define the transformation; note that this is lazy.
            lazy_rdd = rdd.map(lambda x: pipeline.function(x, config=pipeline.config))
            results = lazy_rdd.collect()  # <-- Trigger computation immediately
            if callbacks is not None:
                for result in results:
                    handle_callbacks(callbacks, result, {"name": name})

[tomwhite]

@songhan89 you could do this by transforming the Cubed DAG (a NetworkX MultiDiGraph of pipeline objects) into a Spark DAG of RDD objects, then computing the DAG of RDDs in a single call to collect. This is basically what happens in the BeamExecutor, where PCollection is equivalent to RDD.

Having said that, what you have now is not necessarily a problem. Calling lazy_rdd.collect() does not return the array data back to the client - it will typically be None so there is no issue with memory or anything like that.

There is a problem with callbacks though (which are used for progress bars and other diagnostics), since they will all be fired at the same time for each pipeline node in the DAG. This won't affect the final result, but makes it harder to see what is going on. Perhaps there is a way to retrieve the results progressively as Spark computes the RDD? (Maybe toLocalIterator - but I'm not sure how that's implemented?)

Another thing I wondered: have you tried using a local embedded version of PySpark to run some Cubed unit tests against?

[songhan89]

@tomwhite sorry for the late response. Have not managed to make much progress on your suggestion.

https://github.com/songhan89/cubed/tree/cubed-spark-executor

The unit test result for SparkExecutor is as followed:

Some of the unit test are failing.

cubed/tests/test_array_api.py::test_object_bool[spark] Error occurred during initialization of VM
Too small maximum heap
FAILED
cubed/tests/test_array_api.py::test_arange[spark] Error occurred during initialization of VM
Too small maximum heap
FAILED
cubed/tests/test_array_api.py::test_asarray[spark] Error occurred during initialization of VM
Too small maximum heap
FAILED
cubed/tests/test_array_api.py::test_ones[spark] Error occurred during initialization of VM
Too small maximum heap
FAILED
cubed/tests/test_array_api.py::test_ones_like[spark] Error occurred during initialization of VM
Too small maximum heap
FAILED
cubed/tests/test_array_api.py::test_astype[spark] Error occurred during initialization of VM
Too small maximum heap
FAILED
cubed/tests/test_array_api.py::test_add_top_level_namespace[spark] Error occurred during initialization of VM
Too small maximum heap
FAILED
cubed/tests/test_array_api.py::test_add_different_chunks[spark] Error occurred during initialization of VM
Too small maximum heap
FAILED
cubed/tests/test_array_api.py::test_add_different_chunks_fail[spark] Error occurred during initialization of VM
Too small maximum heap
FAILED
cubed/tests/test_array_api.py::test_negative[spark] Error occurred during initialization of VM
Too small maximum heap
FAILED
cubed/tests/test_array_api.py::test_matmul[spark] Error occurred during initialization of VM
Too small maximum heap
FAILED
cubed/tests/test_array_api.py::test_matmul_cloud[spark] SKIPPED (need --runcloud option to
run)
cubed/tests/test_array_api.py::test_outer[spark] Error occurred during initialization of VM
Too small maximum heap
FAILED
cubed/tests/test_array_api.py::test_broadcast_arrays[spark] 25/02/23 04:08:55 WARN Utils: Your hostname, codespaces-a16525 resolves to a loopback address: 127.0.0.1; using 10.0.0.124 instead (on interface eth0)
25/02/23 04:08:55 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
25/02/23 04:08:56 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
PASSED                                                                          
cubed/tests/test_array_api.py::test_broadcast_to[spark-shape0-chunks0-new_shape0-None-new_chunks_expected0] PASSED
cubed/tests/test_array_api.py::test_broadcast_to[spark-shape1-chunks1-new_shape1-None-new_chunks_expected1] PASSED
cubed/tests/test_array_api.py::test_broadcast_to[spark-shape2-chunks2-new_shape2-None-new_chunks_expected2] PASSED
cubed/tests/test_array_api.py::test_broadcast_to[spark-shape3-chunks3-new_shape3-None-new_chunks_expected3] PASSED
cubed/tests/test_array_api.py::test_broadcast_to[spark-shape4-chunks4-new_shape4-new_chunks4-new_chunks_expected4] PASSED
cubed/tests/test_array_api.py::test_concat[spark] 25/02/23 04:09:04 WARN SparkSession: Using an existing Spark session; only runtime SQL configurations will take effect.
PASSED
cubed/tests/test_array_api.py::test_expand_dims[spark] PASSED
cubed/tests/test_array_api.py::test_flip[spark-shape0-chunks0-None] 25/02/23 04:09:05 WARN SparkSession: Using an existing Spark session; only runtime SQL configurations will take effect.
PASSED
cubed/tests/test_array_api.py::test_flip[spark-shape1-chunks1-0] PASSED
cubed/tests/test_array_api.py::test_flip[spark-shape2-chunks2-0] PASSED
cubed/tests/test_array_api.py::test_flip[spark-shape3-chunks3-None] PASSED
cubed/tests/test_array_api.py::test_flip[spark-shape4-chunks4-0] PASSED
cubed/tests/test_array_api.py::test_flip[spark-shape5-chunks5-1] PASSED
cubed/tests/test_array_api.py::test_flip[spark-shape6-chunks6-axis6] PASSED
cubed/tests/test_array_api.py::test_flip[spark-shape7-chunks7--1] PASSED
cubed/tests/test_array_api.py::test_flip[spark-shape8-chunks8-axis8] PASSED
cubed/tests/test_array_api.py::test_permute_dims[spark] 25/02/23 04:09:09 WARN SparkSession: Using an existing Spark session; only runtime SQL configurations will take effect.
PASSED
cubed/tests/test_array_api.py::test_reshape[spark] PASSED
cubed/tests/test_array_api.py::test_reshape_chunks[spark] PASSED
cubed/tests/test_array_api.py::test_reshape_chunks_with_smaller_end_chunk[spark] PASSED
cubed/tests/test_array_api.py::test_roll[spark-chunks0-3-None] PASSED
cubed/tests/test_array_api.py::test_roll[spark-chunks1-3-0] PASSED
cubed/tests/test_array_api.py::test_roll[spark-chunks2-shift2-axis2] 25/02/23 04:09:12 WARN GarbageCollectionMetrics: To enable non-built-in garbage collector(s) List(G1 Concurrent GC), users should configure it(them) to spark.eventLog.gcMetrics.youngGenerationGarbageCollectors or spark.eventLog.gcMetrics.oldGenerationGarbageCollectors
PASSED
cubed/tests/test_array_api.py::test_roll[spark-chunks3-shift3-None] PASSED
cubed/tests/test_array_api.py::test_roll[spark-chunks4-shift4-1] PASSED
cubed/tests/test_array_api.py::test_squeeze_1d[spark] PASSED
cubed/tests/test_array_api.py::test_squeeze_2d[spark] PASSED
cubed/tests/test_array_api.py::test_stack[spark] PASSED
cubed/tests/test_array_api.py::test_unstack[spark-chunks0] PASSED
cubed/tests/test_array_api.py::test_unstack[spark-chunks1] PASSED
cubed/tests/test_array_api.py::test_unstack[spark-chunks2] PASSED
cubed/tests/test_array_api.py::test_mean_axis_0[spark] PASSED
cubed/tests/test_array_api.py::test_sum[spark] PASSED
cubed/tests/test_array_api.py::test_sum_axis_0[spark] PASSED
cubed/tests/test_array_api.py::test_all[spark] PASSED
cubed/tests/test_array_api.py::test_all_zero_dimension[spark] 25/02/23 04:09:19 WARN SparkSession: Using an existing Spark session; only runtime SQL configurations will take effect.
PASSED
cubed/tests/test_array_api.py::test_object_transpose PASSED
cubed/tests/test_array_api.py::test_reflected_operation PASSED
cubed/tests/test_array_api.py::test_arange_step PASSED
cubed/tests/test_array_api.py::test_asarray_from_array PASSED
cubed/tests/test_array_api.py::test_eye[-1] PASSED
cubed/tests/test_array_api.py::test_eye[0] PASSED
cubed/tests/test_array_api.py::test_eye[1] PASSED
cubed/tests/test_array_api.py::test_linspace[True] PASSED
cubed/tests/test_array_api.py::test_linspace[False] PASSED
cubed/tests/test_array_api.py::test_tril_triu[-1] PASSED
cubed/tests/test_array_api.py::test_tril_triu[0] PASSED
cubed/tests/test_array_api.py::test_tril_triu[1] PASSED
cubed/tests/test_array_api.py::test_add[single-threaded] PASSED
cubed/tests/test_array_api.py::test_add[threads] PASSED
cubed/tests/test_array_api.py::test_add[processes] PASSED
cubed/tests/test_array_api.py::test_add[spark] 25/02/23 04:09:20 WARN SparkSession: Using an existing Spark session; only runtime SQL configurations will take effect.
PASSED
cubed/tests/test_array_api.py::test_add_scalars PASSED
cubed/tests/test_array_api.py::test_clip[None-None] PASSED
cubed/tests/test_array_api.py::test_clip[4-None] PASSED
cubed/tests/test_array_api.py::test_clip[None-7] PASSED
cubed/tests/test_array_api.py::test_clip[4-7] PASSED
cubed/tests/test_array_api.py::test_clip[0-10] PASSED
cubed/tests/test_array_api.py::test_equal PASSED
cubed/tests/test_array_api.py::test_index_1d[ind0] PASSED
cubed/tests/test_array_api.py::test_index_1d[6] PASSED
cubed/tests/test_array_api.py::test_index_1d[ind2] PASSED
cubed/tests/test_array_api.py::test_index_1d[ind3] PASSED
cubed/tests/test_array_api.py::test_index_1d[ind4] PASSED
cubed/tests/test_array_api.py::test_index_1d[ind5] PASSED
cubed/tests/test_array_api.py::test_index_1d[ind6] PASSED
cubed/tests/test_array_api.py::test_index_1d[ind7] PASSED
cubed/tests/test_array_api.py::test_index_1d[ind8] PASSED
cubed/tests/test_array_api.py::test_index_1d[ind9] PASSED
cubed/tests/test_array_api.py::test_index_2d[ind0] PASSED
cubed/tests/test_array_api.py::test_index_2d[ind1] PASSED
cubed/tests/test_array_api.py::test_index_2d[ind2] PASSED
cubed/tests/test_array_api.py::test_index_2d[ind3] PASSED
cubed/tests/test_array_api.py::test_index_2d[ind4] PASSED
cubed/tests/test_array_api.py::test_index_2d[ind5] PASSED
cubed/tests/test_array_api.py::test_index_2d[ind6] PASSED
cubed/tests/test_array_api.py::test_index_2d[ind7] PASSED
cubed/tests/test_array_api.py::test_index_2d[ind8] PASSED
cubed/tests/test_array_api.py::test_index_2d[ind9] PASSED
cubed/tests/test_array_api.py::test_index_2d[ind10] PASSED
cubed/tests/test_array_api.py::test_index_2d[ind11] PASSED
cubed/tests/test_array_api.py::test_index_2d[ind12] PASSED
cubed/tests/test_array_api.py::test_index_2d[ind13] PASSED
cubed/tests/test_array_api.py::test_index_2d[ind14] PASSED
cubed/tests/test_array_api.py::test_index_2d[ind15] PASSED
cubed/tests/test_array_api.py::test_index_2d_no_op[ind0] PASSED
cubed/tests/test_array_api.py::test_index_2d_no_op[ind1] PASSED
cubed/tests/test_array_api.py::test_index_2d_no_op[ind2] PASSED
cubed/tests/test_array_api.py::test_index_2d_no_op[ind3] PASSED
cubed/tests/test_array_api.py::test_index_2d_no_op[ind4] PASSED
cubed/tests/test_array_api.py::test_index_1d_step[20-4-ind0-new_chunks_expected0] PASSED
cubed/tests/test_array_api.py::test_index_1d_step[20-5-ind1-new_chunks_expected1] PASSED
cubed/tests/test_array_api.py::test_index_1d_step[20-8-ind2-new_chunks_expected2] PASSED
cubed/tests/test_array_api.py::test_index_1d_step[50-5-ind3-new_chunks_expected3] PASSED
cubed/tests/test_array_api.py::test_index_2d_step[shape0-chunks0-ind0-new_chunks_expected0] PASSED
cubed/tests/test_array_api.py::test_index_slice_unsupported_step PASSED
cubed/tests/test_array_api.py::test_take[0] PASSED
cubed/tests/test_array_api.py::test_take[1] PASSED
cubed/tests/test_array_api.py::test_matmul_modal[modal_executor0] SKIPPED (got empty parameter
set ['modal_executor'], function test_matmul_modal at /workspaces/codespace...)
cubed/tests/test_array_api.py::test_tensordot[1] PASSED
cubed/tests/test_array_api.py::test_tensordot[axes1] PASSED
cubed/tests/test_array_api.py::test_concat_different_chunks PASSED
cubed/tests/test_array_api.py::test_concat_single_array[None] PASSED
cubed/tests/test_array_api.py::test_concat_single_array[0] PASSED
cubed/tests/test_array_api.py::test_concat_incompatible_shapes PASSED
cubed/tests/test_array_api.py::test_moveaxis PASSED
cubed/tests/test_array_api.py::test_repeat PASSED
cubed/tests/test_array_api.py::test_tile[repetitions0] PASSED
cubed/tests/test_array_api.py::test_tile[repetitions1] PASSED
cubed/tests/test_array_api.py::test_tile[repetitions2] PASSED
cubed/tests/test_array_api.py::test_unstack_zero_arrays PASSED
cubed/tests/test_array_api.py::test_unstack_single_array PASSED
cubed/tests/test_array_api.py::test_argmax PASSED
cubed/tests/test_array_api.py::test_argmax_axis_0 PASSED
cubed/tests/test_array_api.py::test_argmin_axis_0 PASSED
cubed/tests/test_array_api.py::test_searchsorted[left-x10-1-x20-1] PASSED
cubed/tests/test_array_api.py::test_searchsorted[left-x11-1-x21-1] PASSED
cubed/tests/test_array_api.py::test_searchsorted[left-x12-3-x22-2] PASSED
cubed/tests/test_array_api.py::test_searchsorted[left-x13-3-x23-5] PASSED
cubed/tests/test_array_api.py::test_searchsorted[left-x14-3-x24-5] PASSED
cubed/tests/test_array_api.py::test_searchsorted[left-x15-3-x25-x2_chunks5] PASSED
cubed/tests/test_array_api.py::test_searchsorted[right-x10-1-x20-1] PASSED
cubed/tests/test_array_api.py::test_searchsorted[right-x11-1-x21-1] PASSED
cubed/tests/test_array_api.py::test_searchsorted[right-x12-3-x22-2] PASSED
cubed/tests/test_array_api.py::test_searchsorted[right-x13-3-x23-5] PASSED
cubed/tests/test_array_api.py::test_searchsorted[right-x14-3-x24-5] PASSED
cubed/tests/test_array_api.py::test_searchsorted[right-x15-3-x25-x2_chunks5] PASSED
cubed/tests/test_array_api.py::test_searchsorted_sorter_not_implemented PASSED
cubed/tests/test_array_api.py::test_where_scalars PASSED
cubed/tests/test_array_api.py::test_var[False-0.0-None] PASSED
cubed/tests/test_array_api.py::test_var[False-0.0-0] PASSED
cubed/tests/test_array_api.py::test_var[False-0.0-1] PASSED
cubed/tests/test_array_api.py::test_var[False-0.0-axis3] PASSED
cubed/tests/test_array_api.py::test_var[False-1.0-None] PASSED
cubed/tests/test_array_api.py::test_var[False-1.0-0] PASSED
cubed/tests/test_array_api.py::test_var[False-1.0-1] PASSED
cubed/tests/test_array_api.py::test_var[False-1.0-axis3] PASSED
cubed/tests/test_array_api.py::test_var[True-0.0-None] PASSED
cubed/tests/test_array_api.py::test_var[True-0.0-0] PASSED
cubed/tests/test_array_api.py::test_var[True-0.0-1] PASSED
cubed/tests/test_array_api.py::test_var[True-0.0-axis3] PASSED
cubed/tests/test_array_api.py::test_var[True-1.0-None] PASSED
cubed/tests/test_array_api.py::test_var[True-1.0-0] PASSED
cubed/tests/test_array_api.py::test_var[True-1.0-1] PASSED
cubed/tests/test_array_api.py::test_var[True-1.0-axis3] PASSED
cubed/tests/test_array_api.py::test_std[False-0.0-None] PASSED
cubed/tests/test_array_api.py::test_std[False-0.0-0] PASSED
cubed/tests/test_array_api.py::test_std[False-0.0-1] PASSED
cubed/tests/test_array_api.py::test_std[False-0.0-axis3] PASSED
cubed/tests/test_array_api.py::test_std[False-1.0-None] PASSED
cubed/tests/test_array_api.py::test_std[False-1.0-0] PASSED
cubed/tests/test_array_api.py::test_std[False-1.0-1] PASSED
cubed/tests/test_array_api.py::test_std[False-1.0-axis3] PASSED
cubed/tests/test_array_api.py::test_std[True-0.0-None] PASSED
cubed/tests/test_array_api.py::test_std[True-0.0-0] PASSED
cubed/tests/test_array_api.py::test_std[True-0.0-1] PASSED
cubed/tests/test_array_api.py::test_std[True-0.0-axis3] PASSED
cubed/tests/test_array_api.py::test_std[True-1.0-None] PASSED
cubed/tests/test_array_api.py::test_std[True-1.0-0] PASSED
cubed/tests/test_array_api.py::test_std[True-1.0-1] PASSED
cubed/tests/test_array_api.py::test_std[True-1.0-axis3] PASSED
cubed/tests/test_array_api.py::test_var__poorly_conditioned PASSED
cubed/tests/test_core.py::test_as_array_fails PASSED
cubed/tests/test_core.py::test_regular_chunks PASSED
cubed/tests/test_core.py::test_from_array[x0-chunks0-None] PASSED
cubed/tests/test_core.py::test_from_array[x1-chunks1-True] PASSED
cubed/tests/test_core.py::test_from_array[x2--1-True] PASSED
cubed/tests/test_core.py::test_from_array[x3-1-None] PASSED
cubed/tests/test_core.py::test_from_array_zarr PASSED
cubed/tests/test_core.py::test_from_zarr[single-threaded-None] PASSED
cubed/tests/test_core.py::test_from_zarr[single-threaded-sub] PASSED
cubed/tests/test_core.py::test_from_zarr[single-threaded-sub/group] PASSED
cubed/tests/test_core.py::test_store[single-threaded] PASSED
cubed/tests/test_core.py::test_store_multiple[single-threaded] PASSED
cubed/tests/test_core.py::test_store_fails[single-threaded] PASSED
cubed/tests/test_core.py::test_to_zarr[single-threaded-None] PASSED
cubed/tests/test_core.py::test_to_zarr[single-threaded-sub] PASSED
cubed/tests/test_core.py::test_to_zarr[single-threaded-sub/group] PASSED
cubed/tests/test_core.py::test_map_blocks_with_kwargs[single-threaded] PASSED
cubed/tests/test_core.py::test_map_blocks_with_block_id[single-threaded] PASSED
cubed/tests/test_core.py::test_map_blocks_no_array_args[single-threaded] PASSED
cubed/tests/test_core.py::test_multiple_ops[single-threaded] PASSED
cubed/tests/test_core.py::test_rechunk[single-threaded-new_chunks0-expected_chunks0] PASSED
cubed/tests/test_core.py::test_rechunk[single-threaded-new_chunks1-expected_chunks1] PASSED
cubed/tests/test_core.py::test_rechunk[single-threaded-new_chunks2-expected_chunks2] PASSED
cubed/tests/test_core.py::test_rechunk[single-threaded-new_chunks3-expected_chunks3] PASSED
cubed/tests/test_core.py::test_compute_is_idempotent[single-threaded] PASSED
cubed/tests/test_core.py::test_default_spec[single-threaded] PASSED
cubed/tests/test_core.py::test_reduction_multiple_rounds[single-threaded] PASSED
cubed/tests/test_core.py::test_array_pickle[single-threaded] PASSED
cubed/tests/test_core.py::test_measure_reserved_mem[single-threaded] SKIPPED (single-threaded
executor does not support measure_reserved_mem)
cubed/tests/test_core.py::test_from_zarr[threads-None] PASSED
cubed/tests/test_core.py::test_from_zarr[threads-sub] PASSED
cubed/tests/test_core.py::test_from_zarr[threads-sub/group] PASSED
cubed/tests/test_core.py::test_store[threads] PASSED
cubed/tests/test_core.py::test_store_multiple[threads] PASSED
cubed/tests/test_core.py::test_store_fails[threads] PASSED
cubed/tests/test_core.py::test_to_zarr[threads-None] PASSED
cubed/tests/test_core.py::test_to_zarr[threads-sub] PASSED
cubed/tests/test_core.py::test_to_zarr[threads-sub/group] PASSED
cubed/tests/test_core.py::test_map_blocks_with_kwargs[threads] PASSED
cubed/tests/test_core.py::test_map_blocks_with_block_id[threads] PASSED
cubed/tests/test_core.py::test_map_blocks_no_array_args[threads] PASSED
cubed/tests/test_core.py::test_multiple_ops[threads] PASSED
cubed/tests/test_core.py::test_rechunk[threads-new_chunks0-expected_chunks0] PASSED
cubed/tests/test_core.py::test_rechunk[threads-new_chunks1-expected_chunks1] PASSED
cubed/tests/test_core.py::test_rechunk[threads-new_chunks2-expected_chunks2] PASSED
cubed/tests/test_core.py::test_rechunk[threads-new_chunks3-expected_chunks3] PASSED
cubed/tests/test_core.py::test_compute_is_idempotent[threads] PASSED
cubed/tests/test_core.py::test_default_spec[threads] PASSED
cubed/tests/test_core.py::test_reduction_multiple_rounds[threads] PASSED
cubed/tests/test_core.py::test_array_pickle[threads] PASSED
cubed/tests/test_core.py::test_measure_reserved_mem[threads] SKIPPED (threads executor does
not support measure_reserved_mem)
cubed/tests/test_core.py::test_from_zarr[processes-None] PASSED
cubed/tests/test_core.py::test_from_zarr[processes-sub] PASSED
cubed/tests/test_core.py::test_from_zarr[processes-sub/group] PASSED
cubed/tests/test_core.py::test_store[processes] PASSED
cubed/tests/test_core.py::test_store_multiple[processes] PASSED
cubed/tests/test_core.py::test_store_fails[processes] PASSED
cubed/tests/test_core.py::test_to_zarr[processes-None] PASSED
cubed/tests/test_core.py::test_to_zarr[processes-sub] PASSED
cubed/tests/test_core.py::test_to_zarr[processes-sub/group] PASSED
cubed/tests/test_core.py::test_map_blocks_with_kwargs[processes] PASSED
cubed/tests/test_core.py::test_map_blocks_with_block_id[processes] PASSED
cubed/tests/test_core.py::test_map_blocks_no_array_args[processes] PASSED
cubed/tests/test_core.py::test_multiple_ops[processes] PASSED
cubed/tests/test_core.py::test_rechunk[processes-new_chunks0-expected_chunks0] PASSED
cubed/tests/test_core.py::test_rechunk[processes-new_chunks1-expected_chunks1] PASSED
cubed/tests/test_core.py::test_rechunk[processes-new_chunks2-expected_chunks2] PASSED
cubed/tests/test_core.py::test_rechunk[processes-new_chunks3-expected_chunks3] PASSED
cubed/tests/test_core.py::test_compute_is_idempotent[processes] PASSED
cubed/tests/test_core.py::test_default_spec[processes] PASSED
cubed/tests/test_core.py::test_reduction_multiple_rounds[processes] PASSED
cubed/tests/test_core.py::test_array_pickle[processes] PASSED
cubed/tests/test_core.py::test_measure_reserved_mem[processes] PASSED
cubed/tests/test_core.py::test_from_zarr[spark-None] PASSED
cubed/tests/test_core.py::test_from_zarr[spark-sub] PASSED
cubed/tests/test_core.py::test_from_zarr[spark-sub/group] PASSED
cubed/tests/test_core.py::test_store[spark] PASSED
cubed/tests/test_core.py::test_store_multiple[spark] PASSED
cubed/tests/test_core.py::test_store_fails[spark] PASSED
cubed/tests/test_core.py::test_to_zarr[spark-None] PASSED
cubed/tests/test_core.py::test_to_zarr[spark-sub] PASSED
cubed/tests/test_core.py::test_to_zarr[spark-sub/group] PASSED
cubed/tests/test_core.py::test_map_blocks_with_kwargs[spark] PASSED
cubed/tests/test_core.py::test_map_blocks_with_block_id[spark] 25/02/23 04:09:40 WARN SparkSession: Using an existing Spark session; only runtime SQL configurations will take effect.
25/02/23 04:09:41 WARN SparkSession: Using an existing Spark session; only runtime SQL configurations will take effect.
PASSED
cubed/tests/test_core.py::test_map_blocks_no_array_args[spark] PASSED
cubed/tests/test_core.py::test_multiple_ops[spark] PASSED
cubed/tests/test_core.py::test_rechunk[spark-new_chunks0-expected_chunks0] PASSED
cubed/tests/test_core.py::test_rechunk[spark-new_chunks1-expected_chunks1] PASSED
cubed/tests/test_core.py::test_rechunk[spark-new_chunks2-expected_chunks2] PASSED
cubed/tests/test_core.py::test_rechunk[spark-new_chunks3-expected_chunks3] PASSED
cubed/tests/test_core.py::test_compute_is_idempotent[spark] PASSED
cubed/tests/test_core.py::test_default_spec[spark] 25/02/23 04:09:44 WARN SparkSession: Using an existing Spark session; only runtime SQL configurations will take effect.
PASSED
cubed/tests/test_core.py::test_reduction_multiple_rounds[spark] 25/02/23 04:09:45 WARN SparkSession: Using an existing Spark session; only runtime SQL configurations will take effect.
PASSED
cubed/tests/test_core.py::test_array_pickle[spark] 25/02/23 04:09:46 WARN SparkSession: Using an existing Spark session; only runtime SQL configurations will take effect.
PASSED
cubed/tests/test_core.py::test_measure_reserved_mem[spark] SKIPPED (spark executor does not
support measure_reserved_mem)
cubed/tests/test_core.py::test_map_blocks_with_different_block_shapes PASSED
cubed/tests/test_core.py::test_map_blocks_drop_axis_chunking PASSED
cubed/tests/test_core.py::test_map_blocks_with_non_cubed_array PASSED
cubed/tests/test_core.py::test_rechunk_same_chunks PASSED
cubed/tests/test_core.py::test_rechunk_intermediate PASSED
cubed/tests/test_core.py::test_rechunk_merge_chunks_optimization PASSED
cubed/tests/test_core.py::test_default_spec_allowed_mem_exceeded PASSED
cubed/tests/test_core.py::test_default_spec_config_override Terminated

[tomwhite]

@songhan89 Great progress!

It looks like the failures are because some of the Cubed unit tests use a small allowed_mem setting (100000 - i.e. 100kB), and Spark doesn't allow values this small (Too small maximum heap). Do you know what the smallest value that Spark allows?

We could change the Cubed tests in the cases where we don't really care about the allowed_mem setting. For example, in test_arange it would be fine to drop the spec parameter from the test function args, and also from the call to arange, since it would the just pick up the default allowed_mem (2GB), like this:

-def test_arange(spec, executor):
-    a = xp.arange(12, chunks=(5,), spec=spec)
+def test_arange(executor):
+    a = xp.arange(12, chunks=(5,))

You should be able to get most of the test suite working with changes like this.

There are a few tests where the precise memory settings are important since they are testing something about how Cubed calculates memory, but we can tackle those separately.