Tiago de Freitas Pereira
--- a/bob/pipelines/distributed/sge.py
+++ b/bob/pipelines/distributed/sge.py

        self.adapt(minimum=min_jobs, maximum=max_jobs, wait_count=5, interval=120)

+    def get_sge_resources(self):
+        """
+        Get the available resources from `SGEMultipleQueuesCluster.sge_job_spec`.
+        This is useful when it's necessary to set the resource available for 
+        `.compute` method.
+        Check https://distributed.dask.org/en/latest/resources.html#resources-with-collections for more information
+
+        Example
+        -------
+        >>> cluster = SGEMultipleQueuesCluster(sge_job_spec=Q_1DAY_GPU_SPEC) # doctest: +SKIP
+        >>> client = Client(cluster) # doctest: +SKIP
+        >>> my_delayed_task.compute(scheduler=client, resources=resources) # doctest: +SKIP
--- a/bob/pipelines/wrappers.py
+++ b/bob/pipelines/wrappers.py
    """

    def __init__(
-        self,
-        estimator,
-        fit_tag=None,
-        transform_tag=None,
-        **kwargs,
+        self, estimator, fit_tag=None, transform_tag=None, **kwargs,
--- a/doc/python/pipeline_example_dask.py
+++ b/doc/python/pipeline_example_dask.py
-
 # Creating X
 X = numpy.zeros((2, 2))
 # Wrapping X with Samples
 X_as_sample = [Sample(X, key=str(i), metadata=1) for i in range(10)]

 # Building an arbitrary pipeline
-pipeline = make_pipeline(
-    MyBoostedTransformer(
-        transform_extra_arguments=(("metadata", "metadata"),),
-        features_dir="./checkpoint_1",
-    ),
-    MyBoostedFitTransformer(
-        features_dir="./checkpoint_2", model_path="./checkpoint_2/model.pkl",
-    ),
+model_path = "~/dask_tmp"
--- a/doc/python/pipeline_example_dask.py
+++ b/doc/python/pipeline_example_dask.py
-        transform_extra_arguments=(("metadata", "metadata"),),
-        features_dir="./checkpoint_1",
-    ),
-    MyBoostedFitTransformer(
-        features_dir="./checkpoint_2", model_path="./checkpoint_2/model.pkl",
-    ),
+model_path = "~/dask_tmp"
+os.makedirs(model_path, exist_ok=True)
+pipeline = make_pipeline(MyTransformer(), MyFitTranformer())
+
+# Wrapping with sample, checkpoint and dask
+pipeline = bob.pipelines.wrap(
+    ["sample", "checkpoint", "dask"],
+    pipeline,
+    model_path=model_path,
+    transform_extra_arguments=(("metadata", "metadata"),),