Parallel Computation

Civis Platform manages a pool of cloud computing resources. You can access these resources with the tools in the civis.futures and civis.parallel modules.

Warning

civis.parallel is deprecated since civis-python v2.10.0 and will be removed at civis-python v3.0.0 (scheduled for release in February 2027). For parallel computation via Civis Platform, please migrate to CivisFuture, as described in Civis futures below.

Civis Futures

A CivisFuture is a concurrent.futures.Future that tracks a job running in Civis Platform. Because it follows the standard library’s Future API, you can start many Civis Platform jobs at once and collect their results with the usual concurrent.futures helpers, such as concurrent.futures.as_completed() and concurrent.futures.wait().

Fanning out Many Jobs

The following is an example to run multiple Civis Platform jobs and keep track of them while the Python program isn’t blocked and can therefore do other things at the same time. Represent each Civis job as a CivisFuture, keep the futures around, and consume them as they finish. The example starts several Civis Platform container scripts, each computing 2 * num1 + num2 for a different pair of arguments:

import concurrent.futures

import civis

client = civis.APIClient()


def submit(num1, num2):
    """Start a container script and return a CivisFuture for its run."""
    script = client.scripts.post_containers(
        required_resources={"cpu": 512, "memory": 256},
        docker_image_name="civisanalytics/datascience-python",
        docker_command=f"python -c 'print(2 * {num1} + {num2})'",
    )
    return civis.utils.run_job(script.id, client=client)


args = [(0, 1), (1, 1), (2, 1), (3, 1), (4, 1), (5, 1), (6, 1)]

# Map each future back to the arguments it was created with, so that
# out-of-order results can be re-associated with their inputs.
futures = {submit(*a): a for a in args}

# Importantly, this Python code at this juncture is *not* blocked,
# while the Civis Platform jobs launched above are running (some of them
# may finish at any point). If desired for your application,
# you can execute any other Python code before needing to handle these jobs
# via `futures`.

for future in concurrent.futures.as_completed(futures):
    num1, num2 = futures[future]
    run = future.result()  # blocks until this one job is done
    print(f"({num1}, {num2}) finished with state {run.state}")

civis.utils.run_job is a convenience wrapper that starts a run for an existing job and hands back a CivisFuture. concurrent.futures.as_completed() gives you each result as soon as its job finishes, rather than after all jobs have finished. If you want the results in submission order instead, iterate over the futures directly and call .result() on each one.

Passing Inputs and Outputs

CivisFuture does not ship a function, its arguments, or its return value between the parent process and a child job. Instead:

The code you want to run must already exist in the child job’s environment, either baked into the Docker image or cloned from GitHub via the repo_http_uri and repo_ref parameters of post_containers().
You are responsible for moving data. Small inputs can go directly into the docker_command. Larger inputs, and all outputs, should travel through Civis Files.

# A parent job running a child job and retrieving its result.

import io

import civis

client = civis.APIClient()

# Upload the input for one child job.
input_file_id = civis.io.file_to_civis(
    io.BytesIO(b"some input data"), name="my-input", client=client
)

script = client.scripts.post_containers(
    required_resources={"cpu": 512, "memory": 256},
    docker_image_name="civisanalytics/datascience-python",
    docker_command=f"python /app/my_script.py {input_file_id}",
    repo_http_uri="github.com/my-org/my-repo.git",
)
future = civis.utils.run_job(script.id, client=client)
future.result()

# Download what the child job produced.
output_file_id = civis.io.file_id_from_run_output(
    "my-output", future.job_id, future.run_id, client=client
)
buf = io.BytesIO()
civis.io.civis_to_file(output_file_id, buf, client=client)

For the parent to find the output that way, the child job has to register it as a run output. Inside my_script.py, that looks like:

# The child job.

import os

import civis

# Some work has happened and produced an output file.
output_path = "some_output.txt"
client = civis.APIClient()
output_file_id = civis.io.file_to_civis(output_path, name="my-output", client=client)
client.scripts.post_containers_runs_outputs(
    os.environ["CIVIS_JOB_ID"],
    os.environ["CIVIS_RUN_ID"],
    "File",
    output_file_id,
)

As this code shows, a child job can read the CIVIS_JOB_ID and CIVIS_RUN_ID environment variables to identify itself.

You can also pass the parent’s own job and run IDs down to it if the child needs to communicate back to the parent:

# A parent job passing its own job and run IDs to its child job.

import os

import civis

client = civis.APIClient()

# The parent job's own identity, forwarded to the child.
parent_job_id = os.environ["CIVIS_JOB_ID"]
parent_run_id = os.environ["CIVIS_RUN_ID"]

script = client.scripts.post_containers(
    required_resources={"cpu": 512, "memory": 256},
    docker_image_name="civisanalytics/datascience-python",
    docker_command=(
        f"python /app/my_script.py {parent_job_id} {parent_run_id}"
    ),
    repo_http_uri="github.com/my-org/my-repo.git",
)
future = civis.utils.run_job(script.id, client=client)
future.result()

Inside my_script.py, the child job reads those IDs from its arguments and can use them to, for example, attach one of its own outputs to the parent’s run instead of its own:

# A child job that produces some result and sends it back to its parent job
# as a run output at the parent job.

import sys

import civis

parent_job_id, parent_run_id = sys.argv[1], sys.argv[2]

# Some work has happened and produced an output file.
output_path = "some_output.txt"

client = civis.APIClient()
output_file_id = civis.io.file_to_civis(output_path, client=client)
client.scripts.post_containers_runs_outputs(
    parent_job_id,
    parent_run_id,
    "File",
    output_file_id,
)

Errors, Retries, and Concurrency

Calling .result() on a future for a job that failed raises a civis.base.CivisJobFailure. You may catch failures and choose how to handle them:

for future in concurrent.futures.as_completed(futures):
    try:
        run = future.result()
    except Exception as e:
        print(f"Job {future.job_id} failed: {e}")

To stop as soon as anything fails, use concurrent.futures.wait() with return_when=concurrent.futures.FIRST_EXCEPTION, and then cancel the runs that are still going by calling .cancel() on each remaining CivisFuture.

Joblib backend

Warning

civis.parallel, and therefore the Civis joblib backend described in this section, is deprecated since civis-python v2.10.0 and will be removed at civis-python v3.0.0 (scheduled for release in February 2027). For parallel computation via Civis Platform, please migrate to CivisFuture, as described in Civis futures above.

If you can divide your work into multiple independent chunks, each of which takes at least several minutes to run, you can reduce the time your job takes to finish by running each chunk simultaneously in Civis Platform. The Civis joblib backend is a software tool which makes it easier to run many jobs simultaneously.

Things to keep in mind when deciding if the Civis joblib backend is the right tool for your code:

Each function call which is parallelized with the Civis joblib backend will run in a different Civis Platform script. Creating a new script comes with some overhead. It will take between a few seconds and a few minutes for each script to start, depending on whether Civis Platform needs to provision additional resources. If you expect that each function call will complete quickly, instead consider either running them in serial or using extra processes in the same Civis Platform script.
Because function calls run in different scripts, function inputs and outputs must be uploaded to Civis Platform from their origin script and downloaded into their destination. If your functions take very large inputs and/or produce very large outputs, moving the data around will cause additional overhead. Consider either using a different tool or refactoring your code so that the function to be parallelized is no longer moving around large amounts of data.
Some open-source libraries, such as scikit-learn, use joblib to do computations in parallel. If you’re working with such a library, the Civis joblib backend provides an easy way to run these parallel computations in different Civis Platform scripts.

Joblib

joblib is an open source Python library which facilitates parallel processing in Python. While Joblib comes with its own parallel computation tools, it also allows users to define their own “back end”. The Civis Python API client takes advantage of this to let you easily run your own code in parallel through Civis Platform.

The make_backend_factory(), infer_backend_factory(), and make_backend_template_factory() functions allow you to define a “civis” parallel computation backend which will transparently distribute computation in cloud resources managed by Civis Platform.

See the joblib user guide for examples of using joblib to do parallel computation. Note that the descriptions of “memmapping” aren’t relevant to using Civis Platform as a backend, since your jobs will potentially run on different computers and can’t share memory. Using the Civis joblib backend to run jobs in parallel in the cloud looks the same as running jobs in parallel on your local computer, except that you first need to set up the “civis” backend.

How to use

Begin by defining the backend. The Civis joblib backend creates and runs Container Scripts, and the make_backend_factory() function accepts several arguments which will be passed to post_containers(). For example, you could pass a repo_http_uri or repo_ref to clone a repository from GitHub into the container which will run your function. Use the docker_image_name and docker_image_tag to select a custom Docker image for your job. You can provide a setup_cmd to run setup in bash before your function executes in Python. The default setup_cmd will run pip install . in the base directory of any repo_http_uri which you include in your backend setup. Make sure that the environment you define for your Civis backend includes all of the code which your parallel function will call.

The make_backend_factory() function will return a backend factory which should be given to the joblib.register_parallel_backend() function. For example:

>>> from joblib import register_parallel_backend
>>> from civis.parallel import make_backend_factory
>>> be_factory = make_backend_factory()
>>> register_parallel_backend('civis', be_factory)

Direct joblib to use a custom backend by entering a joblib.parallel_config() context:

>>> from joblib import parallel_config
>>> with parallel_config('civis'):
...     # Do joblib parallel computation here.

You can find more about custom joblib backends in the joblib documentation.

Note that joblib.Parallel takes both a n_jobs and pre_dispatch parameter. The Civis joblib backend doesn’t queue submitted jobs itself, so it will run pre_dispatch jobs at once.

Note

Since joblib v1.3.0, n_jobs=1 will disable the use of the specified parallel backend, and will simply run the given function with a regular for loop. n_jobs=1 would be useful for testing your code locally before running it in Civis Platform.

Note

The default value of pre_dispatch is "2*n_jobs", which will run a maximum of 2 * n_jobs jobs at once on Civis Platform. Set pre_dispatch="n_jobs" in your joblib.Parallel call to run at most n_jobs jobs.

The Civis joblib backend uses cloudpickle to transport code and data from the parent environment to Civis Platform. This means that you may parallelize dynamically-defined functions and classes, including lambda functions.

The joblib backend will automatically add environment variables called CIVIS_PARENT_JOB_ID and CIVIS_PARENT_RUN_ID, holding the values of the job and run IDs of the Civis Platform job in which you’re running the joblib backend (if any). Your functions could use these to communicate with the parent job or to recognize that they’re in a process which has been created by another Civis Platform job. However, where possible you should let the joblib backend itself transport the return value of the function it’s running back to the parent.

Infer backend parameters

If you’re writing code which will run inside a Civis Container Script, then the infer_backend_factory() function returns a backend factory with environment parameters pre-populated by inspecting the state of your container script at run time. Use infer_backend_factory() anywhere you would use make_backend_factory(), and you don’t need to specify a Docker image or GitHub repository.

Script Templates

The make_backend_template_factory() is intended for developers who are writing code which may be run by users who don’t have permissions to create new container scripts with the necessary environment.

Instead of defining and creating new container scripts with make_backend_factory(), you can use make_backend_template_factory() to launch custom scripts from a script template. To use the template factory, your backing container script must have the Civis Python client installed, and its run command must finish by calling civis_joblib_worker with no arguments. The template must accept the parameter JOBLIB_FUNC_FILE_ID. The Civis joblib backend will use this parameter to transport your remote work.

Examples

Parallel computation using the default joblib backend (this uses processes on your local computer):

>>> def expensive_calculation(num1, num2):
...     return 2 * num1 + num2
>>> from joblib import delayed, Parallel
>>> parallel = Parallel(n_jobs=5)
>>> args = [(0, 1), (1, 1), (2, 1), (3, 1), (4, 1), (5, 1), (6, 1)]
>>> print(parallel(delayed(expensive_calculation)(*a) for a in args))
[1, 3, 5, 7, 9, 11, 13]

You can do the same parallel computation using the Civis backend by creating and registering a backend factory and entering a with parallel_config('civis') context. The code below will start seven different jobs in Civis Platform (with up to five running at once). Each job will call the function expensive_calculation with a different set of arguments from the list args:

>>> def expensive_calculation(num1, num2):
...     return 2 * num1 + num2
>>> from joblib import delayed, Parallel
>>> from joblib import parallel_config, register_parallel_backend
>>> from civis.parallel import make_backend_factory
>>> register_parallel_backend('civis', make_backend_factory(
...     required_resources={"cpu": 512, "memory": 256}))
>>> args = [(0, 1), (1, 1), (2, 1), (3, 1), (4, 1), (5, 1), (6, 1)]
>>> with parallel_config('civis'):
...     parallel = Parallel(n_jobs=5, pre_dispatch='n_jobs')
...     print(parallel(delayed(expensive_calculation)(*a) for a in args))
[1, 3, 5, 7, 9, 11, 13]

Since joblib v1.4.2, joblib.Parallel has a return_as parameter that can accept 'generator' or 'generator_unordered' (default is 'list', whose behavior is shown in the examples above). Returning a generator, especially the “unordered” version, instead of a list is useful for getting (partial) results back from Civis Platform faster as soon as any child job finishes (as opposed to having to waiting for all child jobs to finish before you get a resulting list). With return_as='generator_unordered', you might want to keep track of the ordering of the child jobs’ results using enumerate():

>>> import inspect
>>> import time
>>> def expensive_calculation(order, num):
...     # lower order for a longer sleep to simulate a longer job
...     time.sleep(10 ** (2 - order))
...     return order, num
>>> from joblib import delayed, Parallel
>>> from joblib import parallel_config, register_parallel_backend
>>> from civis.parallel import make_backend_factory
>>> register_parallel_backend('civis', make_backend_factory(
...     required_resources={"cpu": 512, "memory": 256}))
>>> args = enumerate(['foo', 'bar', 'baz'])  # `enumerate` to get an order index
>>> with parallel_config('civis'):
...     parallel = Parallel(
...         n_jobs=5, pre_dispatch='n_jobs',
...         return_as='generator_unordered',  # yields a result from a child job as soon as it's ready
...     )
...     results = parallel(delayed(expensive_calculation)(*a) for a in args)
...     assert inspect.isgenerator(results)
...     for order, num in results:
...         print(order, num)
2 baz
1 bar
0 foo

Lastly, you can use the Civis joblib backend to parallelize any code which uses joblib internally, such as scikit-learn:

>>> from joblib import parallel_config, register_parallel_backend
>>> from sklearn.model_selection import GridSearchCV
>>> from sklearn.ensemble import GradientBoostingClassifier
>>> from sklearn.datasets import load_digits
>>> digits = load_digits()
>>> param_grid = {
...     "max_depth": [1, 3, 5, None],
...     "max_features": ["sqrt", "log2", None],
...     "learning_rate": [0.1, 0.01, 0.001]
... }
>>> # Note: n_jobs and pre_dispatch specify the maximum number of
>>> # concurrent jobs.
>>> gs = GridSearchCV(GradientBoostingClassifier(n_estimators=1000,
...                                              random_state=42),
...                   param_grid=param_grid,
...                   n_jobs=5, pre_dispatch="n_jobs")
>>> register_parallel_backend('civis', make_backend_factory(
...     required_resources={"cpu": 512, "memory": 256}))
>>> with parallel_config('civis'):
...     gs.fit(digits.data, digits.target)

Debugging

Any (non-retried) errors in child jobs will cause the entire parallel call to fail. joblib will transport the first exception from a remote job and raise it in the parent process so that you can debug.

If your remote jobs are failing because of network problems (e.g., occasional 500 errors), you can make your parallel call more likely to succeed by using a max_job_retries value above 0 when creating your backend factory. This will automatically retry a job (potentially more than once) before giving up and keeping an exception.

Logging: The Civis joblib backend uses the standard library logging module, with debug emits for events which might help you diagnose errors. See also the “verbose” argument to joblib.Parallel, which prints information to either stdout or stderr.

Mismatches between your local environment and the environment in the Civis container script jobs are a common source of errors. To run a function in Civis platform, any modules called by that function must be importable from a Python interpreter running in the container script. For example, if you use joblib.Parallel with numpy.sqrt(), the joblib backend must be set to run your function in a container which has numpy installed. If you see an error such as:

ModuleNotFoundError: No module named 'numpy'

this signifies that the module you’re trying to use doesn’t exist in the remote environment. Select a Docker container with the module installed, or install it in your remote environment by using the repo_http_uri parameter of make_backend_factory() to install it from GitHub.

Object Reference

class civis.futures.CivisFuture(poller: Callable, poller_args: Sequence, polling_interval: int | float | None = None, client: APIClient | None = None, poll_on_creation: bool = True, max_retries: int | None = None)[source]

A class for tracking future results.

This is a subclass of concurrent.futures.Future from the Python standard library. See: https://docs.python.org/3/library/concurrent.futures.html

Parameters:

pollerfunc: A function which returns an object that has a state attribute.
poller_argstuple: The arguments with which to call the poller function.
polling_intervalint or float, optional: The number of seconds between API requests to check whether a result is ready. If an integer or float is provided, this number will be used as the polling interval. If None (the default), the polling interval will start at 1 second and increase geometrically up to 15 seconds. The ratio of the increase is 1.2, resulting in polling intervals in seconds of 1, 1.2, 1.44, 1.728, etc. This default behavior allows for a faster return for a short-running job and a capped polling interval for longer-running jobs.
clientcivis.APIClient, optional
poll_on_creationbool, optional: If True (the default), it will poll upon calling result() the first time. If False, it will wait the number of seconds specified in polling_interval from object creation before polling.
max_retriesint, optional: If the job errors or finishes in a failed state, restart it with an entirely new run, up to this many times. If None or 0 (the default), do not retry. Retries should only be used for idempotent jobs which might fail because of network or other random failures – e.g., do not set this for non-idempotent jobs such as data processes that are not safe to run more than once.

Attributes:

job_id: The job ID for the Civis Platform job that this future is tracking.
job_run_url: The URL for the run of the Civis Platform job that this future is tracking.
job_url: The URL for the Civis Platform job that this future is tracking.
run_id: The run ID for the Civis Platform job that this future is tracking.

Methods

`add_done_callback`(fn)	Attaches a callable that will be called when the future finishes.
`cancel`()	Submit a request to cancel the job/run.
`cancelled`()	Return True if the future was cancelled.
`done`()	Return True if the future was cancelled or finished executing.
`exception`([timeout])	Return the exception raised by the call that the future represents.
`failed`()	Return `True` if the Civis job failed.
`outputs`()	Block on job completion and return a list of run outputs.
`result`([timeout])	Return the result of the call that the future represents.
`running`()	Return True if the future is currently executing.
`set_exception`(exception)	Sets the result of the future as being the given exception.
`set_result`(result)	Sets the return value of work associated with the future.
`set_running_or_notify_cancel`()	Mark the future as running or process any cancel notifications.
`succeeded`()	Return `True` if the job completed in Civis with no error.

cleanup

Examples

This example is provided as a function at query_civis().

>>> import civis
>>> client = civis.APIClient()
>>> database_id = client.get_database_id("my_database")
>>> cred_id = client.default_database_credential_id
>>> sql = "SELECT 1"
>>> preview_rows = 10
>>> response = client.queries.post(database_id, sql, preview_rows,
>>>                                credential=cred_id)
>>>
>>> poller = client.queries.get_runs
>>> poller_args = response.id, response.last_run_id
>>> polling_interval = 10
>>> future = CivisFuture(poller, poller_args, polling_interval)
>>> future.job_id == response.id
True
>>> future.run_id == response.last_run_id
True

cancel() → bool[source]

Submit a request to cancel the job/run.

Returns:

bool: Whether or not the job is in a cancelled state.

property job_id: int

The job ID for the Civis Platform job that this future is tracking.

Returns:

int

property job_run_url: str | None

The URL for the run of the Civis Platform job that this future is tracking.

Returns:

str | None

property job_url: str

The URL for the Civis Platform job that this future is tracking.

Returns:

str

outputs() → list[Response][source]

Block on job completion and return a list of run outputs.

The method will only return run outputs for successful jobs. Failed jobs will raise an exception.

Returns:

list[civis.Response]: List of run outputs from a successfully completed job.

Raises:

civis.base.CivisJobFailure: If the job fails.

property run_id: int | None

The run ID for the Civis Platform job that this future is tracking.

Returns:

int | None

Parallel computations using the Civis Platform infrastructure

Deprecated since version 2.10.0: civis.parallel will be removed at civis-python v3.0.0 (scheduled for release in February 2027). For parallel computation via Civis Platform, please migrate to civis.futures.CivisFuture.

exception civis.parallel.JobSubmissionError[source]: An error occurred while submitting a job to Civis Platform.

civis.parallel.infer_backend_factory(required_resources=None, params=None, arguments=None, client=None, polling_interval=None, setup_cmd=None, max_submit_retries=0, max_job_retries=0, hidden=True, remote_backend='sequential', **kwargs)[source]

Infer the container environment and return a backend factory.

This function helps you run additional jobs from code which executes inside a Civis container job. The function reads settings for relevant parameters (e.g., the Docker image) of the container it’s running inside of.

Jobs created through this backend will have environment variables CIVIS_PARENT_JOB_ID and CIVIS_PARENT_RUN_ID with the contents of the CIVIS_JOB_ID and CIVIS_RUN_ID of the environment which created them. If the code doesn’t have CIVIS_JOB_ID and CIVIS_RUN_ID environment variables available, the child will not have CIVIS_PARENT_JOB_ID and CIVIS_PARENT_RUN_ID environment variables.

Note

This function will read the state of the parent container job at the time this function executes. If the user has modified the container job since the run started (e.g., by changing the GitHub branch in the container’s GUI), this function may infer incorrect settings for the child jobs.

Keyword arguments inferred from the existing script’s state are [‘docker_image_name’, ‘docker_image_tag’, ‘repo_http_uri’, ‘repo_ref’, ‘remote_host_credential_id’, ‘git_credential_id’, ‘cancel_timeout’, ‘time_zone’]

Parameters:

required_resourcesdict or None, optional

The resources needed by the container. See the container scripts API documentation for details. Resource requirements not specified will default to the requirements of the current job.

paramslist or None, optional

A definition of the parameters this script accepts in the arguments field. See the container scripts API documentation for details.

Parameters of the child jobs will default to the parameters of the current job. Any parameters provided here will override parameters of the same name from the current job.

argumentsdict or None, optional

Dictionary of name/value pairs to use to run this script. Only settable if this script has defined params. See the container scripts API documentation for details.

Arguments will default to the arguments of the current job. Anything provided here will override portions of the current job’s arguments.

clientcivis.APIClient instance or None, optional

An API Client object to use.

polling_intervalint, optional

The polling interval, in seconds, for checking container script status. If you have many jobs, you may want to set this higher (e.g., 300) to avoid rate-limiting.

setup_cmdstr, optional

A shell command or sequence of commands for setting up the environment. These will precede the commands used to run functions in joblib. This is primarily for installing dependencies that are not available in the Docker image (e.g., cd /app && pip install . or pip install gensim).

With no GitHub repo input, the setup command will default to a command that does nothing. If a repo_http_uri is provided, the default setup command will attempt to run pip install .. If this command fails, execution will still continue.

max_submit_retriesint, optional

The maximum number of retries for submitting each job. This is to help avoid a large set of jobs failing because of a single 5xx error. A value higher than zero should only be used for jobs that are idempotent (i.e., jobs whose result and side effects are the same regardless of whether they are run once or many times).

max_job_retriesint, optional

Retry failed jobs this number of times before giving up. Even more than with max_submit_retries, this should only be used for jobs which are idempotent, as the job may have caused side effects (if any) before failing. These retries assist with jobs which may have failed because of network or worker failures.

hidden: bool, optional

The hidden status of the object. Setting this to True hides it from most API endpoints. The object can still be queried directly by ID. Defaults to True.

remote_backendstr or object, optional

The name of a joblib backend or a joblib backend itself. This parameter is the joblib backend to use when executing code within joblib in the container. The default of 'sequential' uses the joblib sequential backend in the container. The value ‘civis’ uses an exact copy of the Civis joblib backend that launched the container. Note that with the value 'civis', one can potentially use more jobs than specified by n_jobs.

**kwargs:

Additional keyword arguments will be passed directly to post_containers(), potentially overriding the values of those arguments in the parent environment.

Raises:

RuntimeError: If this function is not running inside a Civis container job.

See also

civis.parallel.make_backend_factory

civis.parallel.make_backend_factory(docker_image_name='civisanalytics/datascience-python', client=None, polling_interval=None, setup_cmd=None, max_submit_retries=0, max_job_retries=0, hidden=True, remote_backend='sequential', **kwargs)[source]

Create a joblib backend factory that uses Civis Container Scripts.

Jobs created through this backend will have environment variables CIVIS_PARENT_JOB_ID and CIVIS_PARENT_RUN_ID with the contents of the CIVIS_JOB_ID and CIVIS_RUN_ID of the environment which created them. If the code doesn’t have CIVIS_JOB_ID and CIVIS_RUN_ID environment variables available, the child will not have CIVIS_PARENT_JOB_ID and CIVIS_PARENT_RUN_ID environment variables.

Note

The total size of function parameters in Parallel() calls on this backend must be less than 5 GB due to AWS file size limits.

Note

The maximum number of concurrent jobs in Civis Platform is controlled by both the n_jobs and pre_dispatch parameters of joblib.Parallel. Set pre_dispatch="n_jobs" to have a maximum of n_jobs processes running at once. (The default is pre_dispatch="2*n_jobs".)

Parameters:

docker_image_namestr, optional

The image for the container script. You may also wish to specify a docker_image_tag in the keyword arguments.

clientcivis.APIClient instance or None, optional

An API Client object to use.

polling_intervalint, optional

The polling interval, in seconds, for checking container script status. If you have many jobs, you may want to set this higher (e.g., 300) to avoid rate-limiting.

setup_cmdstr, optional

A shell command or sequence of commands for setting up the environment. These will precede the commands used to run functions in joblib. This is primarily for installing dependencies that are not available in the dockerhub repo (e.g., cd /app && pip install . or pip install gensim).

With no GitHub repo input, the setup command will default to a command that does nothing. If a repo_http_uri is provided, the default setup command will attempt to run pip install .. If this command fails, execution will still continue.

max_submit_retriesint, optional

The maximum number of retries for submitting each job. This is to help avoid a large set of jobs failing because of a single 5xx error. A value higher than zero should only be used for jobs that are idempotent (i.e., jobs whose result and side effects are the same regardless of whether they are run once or many times).

max_job_retriesint, optional

Retry failed jobs this number of times before giving up. Even more than with max_submit_retries, this should only be used for jobs which are idempotent, as the job may have caused side effects (if any) before failing. These retries assist with jobs which may have failed because of network or worker failures.

hidden: bool, optional

The hidden status of the object. Setting this to true hides it from most API endpoints. The object can still be queried directly by ID. Defaults to True.

remote_backendstr or object, optional

The name of a joblib backend or a joblib backend itself. This parameter is the joblib backend to use when executing code within joblib in the container. The default of 'sequential' uses the joblib sequential backend in the container. The value 'civis' uses an exact copy of the Civis joblib backend that launched the container. Note that with the value 'civis', one can potentially use more jobs than specified by n_jobs.

**kwargs:

Additional keyword arguments will be passed directly to civis.APIClient.scripts.post_containers.

Notes

Joblib’s joblib.parallel.register_parallel_backend() (see example above) expects a callable that returns a joblib.parallel.ParallelBackendBase instance. This function allows the user to specify the Civis container script setting that will be used when that backend creates container scripts to run jobs.

The specified Docker image (optionally, with a GitHub repo and setup command) must have basically the same environment as the one in which this module is used to submit jobs. The worker jobs need to be able to deserialize the jobs they are given, including the data and all the necessary Python objects (e.g., if you pass a Pandas data frame, the image must have Pandas installed). You may use functions and classes dynamically defined in the code (e.g., lambda functions), but if your joblib-parallelized function calls code imported from another module, that module must be installed in the remote environment.

Examples

>>> # Without joblib:
>>> from math import sqrt
>>> print([sqrt(i ** 2) for i in range(10)])
[0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]

>>> # Using the default joblib backend:
>>> from joblib import delayed, Parallel
>>> parallel = Parallel(n_jobs=5)
>>> print(parallel(delayed(sqrt)(i ** 2) for i in range(10)))
[0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]

>>> # Using the Civis backend:
>>> from joblib import parallel_config, register_parallel_backend
>>> from civis.parallel import make_backend_factory
>>> register_parallel_backend('civis', make_backend_factory(
...     required_resources={"cpu": 512, "memory": 256}))
>>> with parallel_config('civis'):
...    parallel = Parallel(n_jobs=5, pre_dispatch='n_jobs')
...    print(parallel(delayed(sqrt)(i ** 2) for i in range(10)))
[0.0, 1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]

>>> # Using scikit-learn with the Civis backend:
>>> from sklearn.model_selection import GridSearchCV
>>> from sklearn.ensemble import GradientBoostingClassifier
>>> from sklearn.datasets import load_digits
>>> digits = load_digits()
>>> param_grid = {
...     "max_depth": [1, 3, 5, None],
...     "max_features": ["sqrt", "log2", None],
...     "learning_rate": [0.1, 0.01, 0.001]
... }
>>> # Note: n_jobs and pre_dispatch specify the maximum number of
>>> # concurrent jobs.
>>> gs = GridSearchCV(GradientBoostingClassifier(n_estimators=1000,
...                                              random_state=42),
...                   param_grid=param_grid,
...                   n_jobs=5, pre_dispatch="n_jobs")
>>> register_parallel_backend('civis', make_backend_factory(
...     required_resources={"cpu": 512, "memory": 256}))
>>> with parallel_config('civis'):
...     gs.fit(digits.data, digits.target)

civis.parallel.make_backend_template_factory(from_template_id, arguments=None, client=None, polling_interval=None, max_submit_retries=0, max_job_retries=0, hidden=True, **kwargs)[source]

Create a joblib backend factory that uses Civis Custom Scripts.

If your template has settable parameters CIVIS_PARENT_JOB_ID and CIVIS_PARENT_RUN_ID, then this executor will fill them with the contents of the CIVIS_JOB_ID and CIVIS_RUN_ID of the environment which created them. If the code doesn’t have CIVIS_JOB_ID and CIVIS_RUN_ID environment variables available, the child will not have CIVIS_PARENT_JOB_ID and CIVIS_PARENT_RUN_ID environment variables.

Parameters:

from_template_id: int: Create jobs as Custom Scripts from the given template ID. When using the joblib backend with templates, the template must have a very specific form. Refer to the documentation for details.
argumentsdict or None, optional: Dictionary of name/value pairs to use to run this script. Only settable if this script has defined params. See the container scripts API documentation for details.
clientcivis.APIClient instance or None, optional: An API Client object to use.
polling_intervalint, optional: The polling interval, in seconds, for checking container script status. If you have many jobs, you may want to set this higher (e.g., 300) to avoid rate-limiting.
max_submit_retriesint, optional: The maximum number of retries for submitting each job. This is to help avoid a large set of jobs failing because of a single 5xx error. A value higher than zero should only be used for jobs that are idempotent (i.e., jobs whose result and side effects are the same regardless of whether they are run once or many times).
max_job_retriesint, optional: Retry failed jobs this number of times before giving up. Even more than with max_submit_retries, this should only be used for jobs which are idempotent, as the job may have caused side effects (if any) before failing. These retries assist with jobs which may have failed because of network or worker failures.
hidden: bool, optional: The hidden status of the object. Setting this to True hides it from most API endpoints. The object can still be queried directly by ID. Defaults to True.
**kwargs:: Additional keyword arguments will be passed directly to civis.APIClient.scripts.post_custom.