SDK

The Jobs class in the Lightning SDK allows you to run batch jobs asynchronously inside a Studio. Use this to offload tasks like data processing, training runs, or inference jobs that do not require interactive execution.

The SDK is automatically available inside Studios. To use the SDK outside a Studio, install it with:

pip install --upgrade lightning-sdk

Ensure you authenticate using environment variables or login via CLI.

export LIGHTNING_USER_ID=your-user-id
export LIGHTNING_API_KEY=your-api-key

Submit a Job

Here is a full example of how to submit a job.

# Install the Lightning SDK
# pip install lightning-sdk

# Export authentication variables
# export LIGHTNING_USER_ID=your-user-id
# export LIGHTNING_API_KEY=your-api-key

from lightning_sdk import Studio, Machine, Job

# Step 1: Initialize a Studio
studio = Studio(name="batch-processing", teamspace="my-teamspace", user="my-user")
studio.start()

# Step 3: Run a Batch Job
job = Job.run(command="python process_data.py", name="data-job", machine=Machine.CPU, studio=studio)

# Step 4: Monitor Job Status
print(job.status)  # Running, Completed, Failed

# Step 5: Stop or Delete a Job
# job.stop()  # Gracefully stop a running job
# job.delete()  # Cancel and remove the job

# Step 6: Shut Down the Studio
studio.stop()

Initialize a Studio

Before running a job, first initialize and start a Studio:

from lightning_sdk import Studio

studio = Studio(name="batch-processing", teamspace="my-teamspace", user="my-user")
studio.start()

Submit the Job

With the Studio defining the compute environment, submit the Job:

job = Job.run(command="python process_data.py", name="data-job", machine=Machine.CPU, studio=studio)

Submit a Job in a private cloud account

Submitting a job on a private own cloud account is as simple specifying the cloud_account argument.

job = Job.run(command="python process_data.py", name="data-job", machine=Machine.CPU, studio=studio, cloud_account="my-cloud-account-name")

List running jobs

teamspace = Teamspace("my-teamspace", user="my-user")

for job in teamspace.jobs:
    print(job.name, job.status)

Check job status

job = Job("data-job", teamspace="my-teamspace", user="my-user")
print(job.status)  # Running, Completed, Failed

Stop or cancel a Job

If a job is no longer needed, you can stop or delete it:

job.stop()  # Gracefully stops a job
job.delete()  # Cancels and removes the job

Run multiple jobs in parallel

You can launch multiple jobs on different machine types:

Job.run(command="python preprocess.py", name="preprocess", machine=Machine.CPU, studio=studio)
Job.run(command="python train.py", name="train-job", machine=Machine.A10G, studio=studio)
Job.run(command="python infer.py", name="inference", machine=Machine.L4

Automate Jobs

GitHub actions

You can schedule jobs using GitHub Actions to run at a fixed time:

name: Scheduled Job

on:
  schedule:
    - cron: '0 3 * * *'  # Runs at 3 AM UTC daily

jobs:
  run_batch:
    runs-on: ubuntu-22.04
    steps:
      - name: Set up Python
        uses: actions/setup-python@v2
        with:
          python-version: "3.10"

      - name: Install Lightning SDK
        run: pip install lightning-sdk

      - name: Start a Job
        run: |
          python -c " \
              from lightning_sdk import Studio, Job, Machine; \
              studio = Studio(name='batch-pipeline', teamspace='my-team', user='ci-bot'); \
              studio.start(); \
              Job.run(command='python daily_task.py', name='daily-task', studio=studio, machine=Machine.A10); \
              studio.stop();"
        env:
          LIGHTNING_USER_ID: ${{ secrets.LIGHTNING_USER_ID }}
          LIGHTNING_API_KEY: ${{ secrets.LIGHTNING_API_KEY }}

Use CPU machines for lightweight preprocessing tasks.
Use GPU machines for deep learning inference and training.
Enable interruptible mode for cost-saving on long-running jobs.
Chain jobs using status checks to create multi-step pipelines.

Job API

init()

The init() method fetches an existing job and accepts the following arguments:

Parameter	Type	Description	Default
name	str	The name of the Job	-
teamspace	str	Teamspace	The teamspace the job is part of
org	str	Organization	The name of the organization owning the teamspace in case it is owned by an org
user	str	User	The name of the user owning the teamspace in case it is owned by a user instead of an org

Example:

job = Job("data-job", teamspace="my-teamspace", user="my-user")

run()

The run() method submits a new job and accepts the following arguments:

Parameter	Type	Description	Default
name	str	The name of the job	-
machine	Machine	Machine to run the Job on	-
command	str	The command to run inside the Studio	None, The command to run inside your job. Required is using a Studio as compute environment. Optional for docker jobs. In case it's not specified for docker jobs, these will run the default command specified by the image.
studio	Studio	str	None
image	str	None	The docker image to run the job with. Mutually exclusive with studio.
teamspace	Teamspace	str	None
org	str	Organization	None
user	str	User	None
cloud_account	str	None	The cloud account to run the job on. Previously this was called cluster.
env	Dict[str, str]	None	Environment variables to set inside the job.
interruptible	bool	Whether the job should run on interruptible instances. They are cheaper but can be preempted.	False
image_credentials	str	None	The credentials used to pull the image. Required if the image is private. This should be the name of the respective credentials secret created on the Lightning AI platform.
cloud_account_auth	bool	Whether to authenticate with the cloud account to pull the image. Required if the registry is part of a cloud provider (e.g. ECR).	False
entrypoint	str	The entrypoint of your docker container. To use the pre-defined entrypoint of the provided image, set this to an empty string. Only applicable when submitting docker jobs.	"sh -c", which just runs the provided command in a standard shell.
path_mappings	Dict[str, str]	None	Dictionary of path mappings. The keys are the path inside the container whereas the value represents the data-connection name and the path inside that connection. Should be of form{ "": ":", "": ""}If the path inside the connection is omitted it's assumed to be the root path of that connection.Only applicable when submitting docker jobs.

Example:

job = Job.run(command="python process_data.py", name="data-job", machine=Machine.T4, studio=studio)

stop()

Stops the respective job. Accepts no arguments.

Example:

job = Job("data-job", teamspace="my-teamspace", org="my-org")
job.stop()

delete()

Deletes the respective job. Note that this also deletes all respective artifacts and is non-reversible. Accepts no arguments.

Example:

job = Job("data-job", teamspace="my-teamspace", org="my-org")
job.delete()

status

Property returning the current status of the job.

Example:

job = Job("data-job", teamspace="my-teamspace", org="my-org")
print(job.status) # -> Status.Pending, Status.Running, Status.Completed, Status.Stopped, Status.Failed

machine

Property returning the machine the job is/was running on.

Example:

job = Job("data-job", teamspace="my-teamspace", org="my-org")
print(job.machine) # -> Machine.A10G

artifacts_uri

Property returning the lit:// address of the job's artifacts in the teamspace drive. It is what lightning ls and lightning cp take, and what the Lightning AI web UI shows under the job.

None when the job keeps no artifacts. A job running a container image only keeps them when it was launched with artifacts_destination, and then they live at that destination rather than under jobs/.

Example:

job = Job("data-job", teamspace="my-teamspace", org="my-org")
print(job.artifacts_uri) # -> "lit://my-org/my-teamspace/jobs/data-job"

A Studio-backed job's files sit directly under jobs/<job name>, both in the drive and at /teamspace/jobs/<job name> in a Studio in the same teamspace, so a path copied out of a Studio addresses the drive as-is.

list_artifacts()

Lists what the job wrote to the teamspace drive. Returns one entry per file or folder, each with a path, is_dir, size and last_modified. Empty when the job keeps no artifacts.

Parameter	Type	Description	Default
path	str	Subfolder of the job's artifacts to list.	"", all of them
recursive	bool	Whether to descend into subfolders.	False

Example:

job = Job("data-job", teamspace="my-teamspace", org="my-org")
for entry in job.list_artifacts(recursive=True):
    if not entry.is_dir:
        print(f"{entry.path} ({entry.size} bytes)")

download_artifacts()

Downloads what the job wrote to a local directory. Raises RuntimeError when the job keeps no artifacts.

Parameter	Type	Description	Default
target_dir	str \| Path	Local directory to download into. Created if it does not exist.	".", the working directory
path	str	Subfolder of the job's artifacts to download.	"", all of them

Example:

job = Job("data-job", teamspace="my-teamspace", org="my-org")
job.download_artifacts("./artifacts")

# Take one subfolder, so a long run does not have to come down whole
job.download_artifacts("./checkpoints", path="checkpoints")

name

Property returning the name of the job.

Example:

job = Job("data-job", teamspace="my-teamspace", org="my-org")
print(job.name) # -> "data-job"

teamspace

Property returning the teamspace the job is in.

Example:

job = Job("data-job", teamspace="my-teamspace", org="my-org")
print(job.teamspace) # -> Teamspace("my-teamspace", org="my-org")

studio

Property returning the studio used as compute environment if it wasn't launched with a docker image.

Example:

job = Job("data-job", teamspace="my-teamspace", org="my-org")
print(job.studio) # -> Studio("my-studio", teamspace="my-teamspace", org="my-org") or None

image

Property returning the image used as compute environment if it wasn't launched with a Studio.

Example:

job = Job("data-job", teamspace="my-teamspace", org="my-org")
print(job.image) # -> "ubuntu:latest" or None

command

Property returning the command ran within the job.

Example:

job = Job("data-job", teamspace="my-teamspace", org="my-org")
print(job.command) # -> "python train.py"

logs

Property returning the logs generated by running the job. Will return an error if the job is still pending or running (check job.status)

Example:

job = Job("data-job", teamspace="my-teamspace", org="my-org")
print(job.logs) # -> "Epoch 1/2: ################"

link

Property returning the link to view the job in the UI.

Example:

job = Job("data-job", teamspace="my-teamspace", org="my-org")
print(job.link) # -> "https://lightning.ai/my-org/my-teamspace/studios/my-studio/app?app_id=jobs&job_name=data-job"

Was this page helpful?

SDK

Submit a Job

Initialize a Studio

Submit the Job

Submit a Job in a private cloud account

List running jobs

Check job status

Stop or cancel a Job

Run multiple jobs in parallel

Automate Jobs

GitHub actions

Job API

__init__()

run()

stop()

delete()

status

machine

artifacts_uri

list_artifacts()

download_artifacts()

name

teamspace

studio

image

command

logs

link

init()