Manage jobs

Introduction

This page covers the Gravitino API for job templates and jobs. For what a template is, how it relates to a run, and what the job statuses mean, see Jobs.

Jobs run through a job executor, set with gravitino.job.executor. The default, local, launches the job as a process on the Gravitino server and is intended for testing. Running jobs anywhere else means implementing an executor. See Custom job executor.

1. The job system is still under development, so some features may not be fully implemented yet.
2. The aim of the job system is not to replace the existing job executors. So, it can only support running a single job at a time, and it doesn't support job scheduling for now.

Job template operations

Register shell

A shell template runs an executable. Placeholders in arguments, environments, and customFields are filled in when a job runs.

Define a shell template

{
  "name": "nightly_export",
  "jobType": "shell",
  "comment": "Exports a table to a drop location",
  "executable": "/opt/jobs/export.sh",
  "arguments": ["{{table}}", "{{target}}"],
  "environments": {"REGION": "{{region}}"},
  "scripts": ["/opt/jobs/lib/common.sh"]
}

executable and scripts must be reachable by the Gravitino server, which accepts local paths and HTTP, HTTPS, FTP, and FTPS URLs.

REST

Register a shell template

curl -X POST -H "Accept: application/vnd.gravitino.v1+json" \
  -H "Content-Type: application/json" -d '{
  "jobTemplate": {
    "name": "nightly_export",
    "jobType": "shell",
    "comment": "Exports a table to a drop location",
    "executable": "/opt/jobs/export.sh",
    "arguments": ["{{table}}", "{{target}}"]
  }
}' http://localhost:8090/api/metalakes/example/jobs/templates

Register Spark

A Spark template submits an application. Running one with the local executor needs either gravitino.jobExecutor.local.sparkHome or SPARK_HOME set before the server starts, or the job fails to launch.

Define a Spark template

{
  "name": "nightly_rollup",
  "jobType": "spark",
  "comment": "Rolls up daily aggregates",
  "executable": "/opt/jobs/rollup.jar",
  "className": "com.example.Rollup",
  "arguments": ["{{date}}"],
  "configs": {"spark.executor.memory": "4g"}
}

List, get, and delete

REST

List, get, and delete templates

curl -X GET -H "Accept: application/vnd.gravitino.v1+json" \
  http://localhost:8090/api/metalakes/example/jobs/templates

curl -X GET -H "Accept: application/vnd.gravitino.v1+json" \
  http://localhost:8090/api/metalakes/example/jobs/templates/nightly_export

curl -X DELETE -H "Accept: application/vnd.gravitino.v1+json" \
  http://localhost:8090/api/metalakes/example/jobs/templates/nightly_export

Java

List, get, and delete templates

List<JobTemplate> templates = client.listJobTemplates();
JobTemplate template = client.getJobTemplate("nightly_export");
boolean deleted = client.deleteJobTemplate("nightly_export");

Python

List, get, and delete templates

templates = client.list_job_templates()
template = client.get_job_template("nightly_export")
deleted = client.delete_job_template("nightly_export")

A template cannot be deleted while jobs from it are queued or running.

Alter

Change	JSON	Java
Rename	{"@type":"rename","newName":"nightly_export_v2"}	JobTemplateChange.rename("nightly_export_v2")
Update the comment	{"@type":"updateComment","newComment":"new_comment"}	JobTemplateChange.updateComment("new_comment")
Update the template	{"@type":"updateTemplate","newTemplate":{...}}	JobTemplateChange.updateTemplate(...)

Job operations

Run

Running names a template and supplies values for its placeholders.

REST

Run a job

curl -X POST -H "Accept: application/vnd.gravitino.v1+json" \
  -H "Content-Type: application/json" -d '{
  "jobTemplateName": "nightly_export",
  "jobConf": {
    "table": "sales.public.orders",
    "target": "s3a://exports/orders",
    "region": "us"
  }
}' http://localhost:8090/api/metalakes/example/jobs/runs

Java

Run a job

JobHandle job = client.runJob(
    "nightly_export",
    ImmutableMap.of(
        "table", "sales.public.orders",
        "target", "s3a://exports/orders",
        "region", "us"));

Python

Run a job

job = client.run_job(
    job_template_name="nightly_export",
    job_conf={
        "table": "sales.public.orders",
        "target": "s3a://exports/orders",
        "region": "us",
    })

List, get, and cancel

Listing can be filtered to one template. A job is identified by its id, and cancelling is a POST to the job's own path.

REST

List, get, and cancel jobs

curl -X GET -H "Accept: application/vnd.gravitino.v1+json" \
  "http://localhost:8090/api/metalakes/example/jobs/runs?jobTemplateName=nightly_export"

curl -X GET -H "Accept: application/vnd.gravitino.v1+json" \
  http://localhost:8090/api/metalakes/example/jobs/runs/{job_id}

curl -X POST -H "Accept: application/vnd.gravitino.v1+json" \
  http://localhost:8090/api/metalakes/example/jobs/runs/{job_id}

Java

List, get, and cancel jobs

List<JobHandle> jobs = client.listJobs("nightly_export");
JobHandle job = client.getJob(jobId);
JobHandle cancelling = client.cancelJob(jobId);

Python

List, get, and cancel jobs

jobs = client.list_jobs(job_template_name="nightly_export")
job = client.get_job(job_id)
cancelling = client.cancel_job(job_id)

Cancelling is a request rather than an instant. The job moves to CANCELLING and then to CANCELLED, and one that finishes first keeps the status it finished with.

Job system configuration

Configure the job system in my-values.yaml under gravitino.job and apply it by re-running the helm upgrade --install command. The chart writes stagingDir, stagingDirKeepTimeInMs, statusPullIntervalInMs and sparkHome from those values, and by default puts the staging directory on the logs volume. Set a property with no chart value under gravitino.additionalConfigItems; see Set other server properties.

my-values.yaml

gravitino:
  job:
    statusPullIntervalInMs: "300000"

The following are the default configurations:

Property name	Description	Default value	Required
gravitino.job.stagingDir	Directory for managing the staging files when running jobs	/tmp/gravitino/jobs/staging	No
gravitino.job.executor	The job executor to use for running jobs	local	No
gravitino.job.stagingDirKeepTimeInMs	The time in milliseconds to keep the staging directory after the job is completed	604800000 (7 days)	No
gravitino.job.statusPullIntervalInMs	The interval in milliseconds to pull the job status from the job executor	300000 (5 minutes)	No

Configurations for local job executor

The local job executor is used for testing and development purposes. It runs the job in the local process. The following are the default configurations for the local job executor:

Property name	Description	Default value	Required
gravitino.jobExecutor.local.waitingQueueSize	The size of the waiting queue for queued jobs in the local job executor	100	No
gravitino.jobExecutor.local.maxRunningJobs	The maximum number of running jobs in the local job executor	max(1, min(available cores / 2, 10))	No
gravitino.jobExecutor.local.jobStatusKeepTimeInMs	The time in milliseconds to keep the job status in the local job executor	3600000 (1 hour)	No
gravitino.jobExecutor.local.sparkHome	The home directory of Spark, Gravitino checks this configuration firstly and then SPARK_HOME env. Either of them should be set to run Spark job	None	No

Future work

The job system still needs more work:

1. Support modification of job templates.
2. Support running Spark jobs (Java and PySpark) based on the Spark job template in the local job executor.
3. Support more job templates, like Python, SQL, etc.
4. Support more job executors, like Apache Airflow, Apache Livy, etc.
5. Support uploading job template related artifacts to the Gravitino server, and also support downloading the artifacts from more distributed file systems like HDFS, S3, etc.
6. Support job scheduling, like running jobs periodically, or based on some events.