diff --git a/cli_tests/app.sh b/cli_tests/app.sh
index b6596f24..d4bce32d 100755
--- a/cli_tests/app.sh
+++ b/cli_tests/app.sh
@@ -17,6 +17,7 @@ setup() {
   FRAMEWORK="myframework"
   APP_IMAGE="gcr.io/shipa-ci/sample-go-app:latest"
   APP_NAME="sample-app"
+  JOB_NAME="sample-job"
   CNAME="my-cname.com"
   TEST_ENVVAR_KEY="FOO"
   TEST_ENVVAR_VALUE="BAR"
@@ -25,6 +26,7 @@ setup() {
 teardown() {
   rm -f app.yaml
   rm -f framework.yaml
+  rm -f job.yaml
 }
 
 @test "help" {
@@ -197,6 +199,47 @@ EOF
   [[ $status -eq 0 ]]
 }
 
+@test "job deploy with yaml file" {
+  cat << EOF > job.yaml
+name: "$JOB_NAME"
+version: v1
+type: Job
+framework: "$FRAMEWORK"
+description: cli test job
+EOF
+  run $KETCH job deploy job.yaml
+  [[ $status -eq 0 ]]
+
+  dataRegex="$JOB_NAME[ \t]+v1[ \t]+$FRAMEWORK[ \t]+cli test job"
+  result=$($KETCH job list $JOB_NAME)
+  echo "RECEIVED:" $result
+  [[ $result =~ $dataRegex ]]
+}
+
+@test "job list" {
+  result=$($KETCH job list)
+  headerRegex="NAME[ \t]+VERSION[ \t]+FRAMEWORK[ \t]+DESCRIPTION"
+  dataRegex="$JOB_NAME[ \t]+v1[ \t]+$FRAMEWORK[ \t]+cli test job"
+  echo "RECEIVED:" $result
+  [[ $result =~ $headerRegex ]]
+  [[ $result =~ $dataRegex ]]
+}
+
+@test "job export" {
+  run $KETCH job export "$JOB_NAME" -f job.yaml
+  result=$(cat job.yaml)
+  echo "RECEIVED:" $result
+  [[ $result =~ "name: $JOB_NAME" ]]
+  [[ $result =~ "type: Job" ]]
+  [[ $result =~ "framework: $FRAMEWORK" ]]
+}
+
+@test "job remove" {
+  result=$($KETCH job remove "$JOB_NAME")
+  echo "RECEIVED:" $result
+  [[ $result =~ "Successfully removed!" ]]
+}
+
 @test "builder list" {
   result=$($KETCH builder list)
   headerRegex="VENDOR[ \t]+IMAGE[ \t]+DESCRIPTION"
diff --git a/cmd/ketch/auto_completion.go b/cmd/ketch/auto_completion.go
index cbb00605..259c411b 100644
--- a/cmd/ketch/auto_completion.go
+++ b/cmd/ketch/auto_completion.go
@@ -87,3 +87,11 @@ func autoCompleteFrameworkNames(cfg config, toComplete ...string) ([]string, cob
 func autoCompleteBuilderNames(cfg config, toComplete ...string) ([]string, cobra.ShellCompDirective) {
 	return builderList.Names(toComplete...), cobra.ShellCompDirectiveNoSpace
 }
+
+func autoCompleteJobNames(cfg config, toComplete ...string) ([]string, cobra.ShellCompDirective) {
+	names, err := jobListNames(cfg, toComplete...)
+	if err != nil {
+		return []string{err.Error()}, cobra.ShellCompDirectiveError
+	}
+	return names, cobra.ShellCompDirectiveNoSpace
+}
diff --git a/cmd/ketch/errors.go b/cmd/ketch/errors.go
index 8bb7f5a7..84de9573 100644
--- a/cmd/ketch/errors.go
+++ b/cmd/ketch/errors.go
@@ -15,6 +15,9 @@ const (
 	ErrInvalidAppName cliError = "invalid app name, app name should have at most 40 " +
 		"characters, containing only lower case letters, numbers or dashes, starting with a letter"
 
+	ErrInvalidJobName cliError = "invalid job name, job name should have at most 40 " +
+		"characters, containing only lower case letters, numbers or dashes, starting with a letter"
+
 	ErrNoEntrypointAndCmd   cliError = "image doesn't have entrypoint and cmd set"
 	ErrLogUnknownTimeFormat cliError = "unknown time format"
 
diff --git a/cmd/ketch/job.go b/cmd/ketch/job.go
new file mode 100644
index 00000000..965c27de
--- /dev/null
+++ b/cmd/ketch/job.go
@@ -0,0 +1,55 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"io"
+
+	"k8s.io/apimachinery/pkg/types"
+
+	ketchv1 "github.com/shipa-corp/ketch/internal/api/v1beta1"
+	"github.com/spf13/cobra"
+)
+
+const jobHelp = `
+Manage jobs.
+`
+
+func newJobCmd(cfg config, out io.Writer) *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "job",
+		Short: "Manage Jobs",
+		Long:  jobHelp,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			return cmd.Usage()
+		},
+	}
+	cmd.AddCommand(newJobListCmd(cfg, out))
+	cmd.AddCommand(newJobDeployCmd(cfg, out))
+	cmd.AddCommand(newJobRemoveCmd(cfg, out))
+	cmd.AddCommand(newJobExportCmd(cfg, out))
+	return cmd
+}
+
+func getJob(ctx context.Context, cfg config, jobName string) (*ketchv1.Job, error) {
+	var frameworks ketchv1.FrameworkList
+	if err := cfg.Client().List(ctx, &frameworks); err != nil {
+		return nil, fmt.Errorf("failed to get frameworks: %w", err)
+	}
+	var namespace string
+	for _, framework := range frameworks.Items {
+		if framework.HasJob(jobName) {
+			namespace = framework.Spec.NamespaceName
+			break
+		}
+	}
+	if namespace == "" {
+		return nil, fmt.Errorf("cannot find framework with job %s", jobName)
+	}
+
+	job := ketchv1.Job{}
+	if err := cfg.Client().Get(ctx, types.NamespacedName{Name: jobName, Namespace: namespace}, &job); err != nil {
+		return nil, fmt.Errorf("failed to get job: %w", err)
+	}
+	return &job, nil
+}
diff --git a/cmd/ketch/job_deploy.go b/cmd/ketch/job_deploy.go
new file mode 100644
index 00000000..bfd078bd
--- /dev/null
+++ b/cmd/ketch/job_deploy.go
@@ -0,0 +1,96 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+
+	"github.com/spf13/cobra"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/yaml"
+
+	ketchv1 "github.com/shipa-corp/ketch/internal/api/v1beta1"
+)
+
+const jobDeployHelp = `
+List all jobs.
+`
+
+const (
+	defaultJobVersion       = "v1"
+	defaultJobParallelism   = 1
+	defaultJobCompletions   = 1
+	defaultJobBackoffLimit  = 6
+	defaultJobRestartPolicy = "Never"
+)
+
+func newJobDeployCmd(cfg config, out io.Writer) *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "deploy [FILENAME]",
+		Short: "Deploy a job.",
+		Long:  jobDeployHelp,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			filename := args[0]
+			return jobDeploy(cmd.Context(), cfg, filename, out)
+		},
+	}
+	return cmd
+}
+
+func jobDeploy(ctx context.Context, cfg config, filename string, out io.Writer) error {
+	b, err := os.ReadFile(filename)
+	if err != nil {
+		return err
+	}
+	var job ketchv1.Job
+	err = yaml.Unmarshal(b, &job.Spec)
+	if err != nil {
+		return err
+	}
+	setJobSpecDefaults(&job.Spec)
+	if err = validateJobSpec(&job.Spec); err != nil {
+		return err
+	}
+	var framework ketchv1.Framework
+	if err := cfg.Client().Get(ctx, types.NamespacedName{Name: job.Spec.Framework}, &framework); err != nil {
+		return fmt.Errorf("failed to get framework: %s", job.Spec.Framework)
+	}
+	job.ObjectMeta.Namespace = framework.Spec.NamespaceName
+	job.ObjectMeta.Name = job.Spec.Name
+	if err := cfg.Client().Create(ctx, &job); err != nil {
+		return fmt.Errorf("failed to create job: %w", err)
+	}
+	fmt.Fprintln(out, "Successfully added!")
+	return nil
+}
+
+// setJobSpecDefaults sets defaults on job.Spec for some unset fields
+func setJobSpecDefaults(jobSpec *ketchv1.JobSpec) {
+	jobSpec.Type = "Job"
+	if jobSpec.Version == "" {
+		jobSpec.Version = defaultJobVersion
+	}
+	if jobSpec.Parallelism == 0 {
+		jobSpec.Parallelism = defaultJobParallelism
+	}
+	if jobSpec.Completions == 0 && jobSpec.Parallelism > 1 {
+		jobSpec.Completions = defaultJobCompletions
+	}
+	if jobSpec.BackoffLimit == 0 {
+		jobSpec.BackoffLimit = defaultJobBackoffLimit
+	}
+	if jobSpec.Policy.RestartPolicy == "" {
+		jobSpec.Policy.RestartPolicy = defaultJobRestartPolicy
+	}
+}
+
+// validateJobSpec assures that required fields are populated. Missing fields will throw errors
+// when the custom resource is created, but this is a way to surface errors to user clearly.
+func validateJobSpec(jobSpec *ketchv1.JobSpec) error {
+	if jobSpec.Name == "" || jobSpec.Framework == "" {
+		return errors.New("job.name and job.framework are required")
+	}
+	return nil
+}
diff --git a/cmd/ketch/job_deploy_test.go b/cmd/ketch/job_deploy_test.go
new file mode 100644
index 00000000..912a9a28
--- /dev/null
+++ b/cmd/ketch/job_deploy_test.go
@@ -0,0 +1,133 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+
+	ketchv1 "github.com/shipa-corp/ketch/internal/api/v1beta1"
+	"github.com/shipa-corp/ketch/internal/mocks"
+	"github.com/shipa-corp/ketch/internal/utils/conversions"
+)
+
+func TestJobDeploy(t *testing.T) {
+	mockFramework := &ketchv1.Framework{ObjectMeta: metav1.ObjectMeta{Name: "myframework"}, Spec: ketchv1.FrameworkSpec{
+		Version:       "v1",
+		NamespaceName: "ketch-myframework",
+		Name:          "myframework",
+		AppQuotaLimit: conversions.IntPtr(1),
+		IngressController: ketchv1.IngressControllerSpec{
+			ClassName:       "traefik",
+			ServiceEndpoint: "10.10.20.30",
+			IngressType:     "traefik",
+			ClusterIssuer:   "letsencrypt",
+		},
+	}}
+	tests := []struct {
+		name        string
+		jobName     string
+		cfg         config
+		filename    string
+		yamlData    string
+		wantJobSpec ketchv1.JobSpec
+		wantOut     string
+		wantErr     string
+	}{
+		{
+			name:    "job from yaml file",
+			jobName: "hello",
+			cfg: &mocks.Configuration{
+				CtrlClientObjects:    []runtime.Object{mockFramework},
+				DynamicClientObjects: []runtime.Object{},
+			},
+			filename: "job.yaml",
+			yamlData: `name: hello
+version: v1
+framework: myframework
+description: test
+parallelism: 1
+completions: 1
+suspend: false
+backoffLimit: 6
+containers: 
+  - name: lister
+    image: ubuntu
+    command:
+      - ls
+      - /
+policy:
+  restartPolicy: Never
+`,
+			wantJobSpec: ketchv1.JobSpec{
+				Name:         "hello",
+				Version:      "v1",
+				Type:         "Job",
+				Framework:    "myframework",
+				Description:  "test",
+				Parallelism:  1,
+				Completions:  1,
+				Suspend:      false,
+				BackoffLimit: 6,
+				Containers: []ketchv1.Container{
+					{
+						Name:    "lister",
+						Image:   "ubuntu",
+						Command: []string{"ls", "/"},
+					},
+				},
+				Policy: ketchv1.Policy{
+					RestartPolicy: "Never",
+				},
+			},
+			wantOut: "Successfully added!\n",
+		},
+		{
+			name:    "error - no framework found",
+			jobName: "hello",
+			cfg: &mocks.Configuration{
+				CtrlClientObjects:    []runtime.Object{mockFramework},
+				DynamicClientObjects: []runtime.Object{},
+			},
+			filename: "job.yaml",
+			yamlData: `name: hello
+version: v1
+framework: NOFRAMEWORK
+description: test
+`,
+			wantErr: "failed to get framework: NOFRAMEWORK",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if tt.yamlData != "" {
+				file, err := os.CreateTemp(t.TempDir(), "*.yaml")
+				require.Nil(t, err)
+				_, err = file.Write([]byte(tt.yamlData))
+				require.Nil(t, err)
+				defer os.Remove(file.Name())
+				tt.filename = file.Name()
+			}
+			out := &bytes.Buffer{}
+			err := jobDeploy(context.Background(), tt.cfg, tt.filename, out)
+			if len(tt.wantErr) > 0 {
+				require.NotNil(t, err)
+				require.Equal(t, tt.wantErr, err.Error())
+				return
+			} else {
+				require.Nil(t, err)
+			}
+			require.Equal(t, tt.wantOut, out.String())
+
+			gotJob := ketchv1.Job{}
+			err = tt.cfg.Client().Get(context.Background(), types.NamespacedName{Name: tt.jobName, Namespace: mockFramework.Spec.NamespaceName}, &gotJob)
+			require.Nil(t, err)
+			require.Equal(t, tt.wantJobSpec, gotJob.Spec)
+		})
+	}
+}
diff --git a/cmd/ketch/job_export.go b/cmd/ketch/job_export.go
new file mode 100644
index 00000000..43406b6e
--- /dev/null
+++ b/cmd/ketch/job_export.go
@@ -0,0 +1,70 @@
+package main
+
+import (
+	"context"
+	"io"
+	"os"
+
+	"sigs.k8s.io/yaml"
+
+	"github.com/shipa-corp/ketch/internal/validation"
+
+	"github.com/spf13/cobra"
+)
+
+const jobExportHelp = `
+Export a job's configuration file.
+`
+
+type jobExportOptions struct {
+	filename string
+	name     string
+}
+
+func newJobExportCmd(cfg config, out io.Writer) *cobra.Command {
+	var options jobExportOptions
+	cmd := &cobra.Command{
+		Use:   "export JOB",
+		Short: "Export a job.",
+		Long:  jobExportHelp,
+		Args:  cobra.ExactValidArgs(1),
+		RunE: func(cmd *cobra.Command, args []string) error {
+			options.name = args[0]
+			if !validation.ValidateName(options.name) {
+				return ErrInvalidJobName
+			}
+			return jobExport(cmd.Context(), cfg, options, out)
+		},
+		ValidArgsFunction: func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
+			return autoCompleteJobNames(cfg, toComplete)
+		},
+	}
+	cmd.Flags().StringVarP(&options.filename, "filename", "f", "", "filename for job export")
+	return cmd
+}
+
+func jobExport(ctx context.Context, cfg config, options jobExportOptions, out io.Writer) error {
+	job, err := getJob(ctx, cfg, options.name)
+	if err != nil {
+		return err
+	}
+	if options.filename != "" {
+		// open file, err if exist, write application
+		_, err := os.Stat(options.filename)
+		if !os.IsNotExist(err) {
+			return errFileExists
+		}
+		f, err := os.Create(options.filename)
+		if err != nil {
+			return err
+		}
+		defer f.Close()
+		out = f
+	}
+	b, err := yaml.Marshal(job.Spec)
+	if err != nil {
+		return err
+	}
+	_, err = out.Write(b)
+	return err
+}
diff --git a/cmd/ketch/job_export_test.go b/cmd/ketch/job_export_test.go
new file mode 100644
index 00000000..c60b44ee
--- /dev/null
+++ b/cmd/ketch/job_export_test.go
@@ -0,0 +1,115 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+
+	ketchv1 "github.com/shipa-corp/ketch/internal/api/v1beta1"
+	"github.com/shipa-corp/ketch/internal/mocks"
+	"github.com/shipa-corp/ketch/internal/utils/conversions"
+)
+
+func TestJobExport(t *testing.T) {
+	mockFramework := &ketchv1.Framework{ObjectMeta: metav1.ObjectMeta{Name: "myframework"}, Spec: ketchv1.FrameworkSpec{
+		Version:       "v1",
+		NamespaceName: "ketch-myframework",
+		Name:          "myframework",
+		AppQuotaLimit: conversions.IntPtr(1),
+		IngressController: ketchv1.IngressControllerSpec{
+			ClassName:       "traefik",
+			ServiceEndpoint: "10.10.20.30",
+			IngressType:     "traefik",
+			ClusterIssuer:   "letsencrypt",
+		},
+	},
+		Status: ketchv1.FrameworkStatus{Jobs: []string{"hello"}}}
+	mockJob := &ketchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{Name: "hello", Namespace: mockFramework.Spec.NamespaceName},
+		Spec: ketchv1.JobSpec{
+			Name:         "hello",
+			Version:      "v1",
+			Framework:    "myframework",
+			Description:  "test",
+			Parallelism:  1,
+			Completions:  1,
+			Suspend:      false,
+			BackoffLimit: 6,
+			Containers: []ketchv1.Container{
+				{
+					Name:    "lister",
+					Image:   "ubuntu",
+					Command: []string{"ls", "/"},
+				},
+			},
+			Policy: ketchv1.Policy{
+				RestartPolicy: "Never",
+			},
+			Type: "Job",
+		},
+	}
+	tests := []struct {
+		name         string
+		cfg          config
+		options      jobExportOptions
+		wantYamlData string
+		wantErr      string
+	}{
+		{
+			name: "success",
+			cfg: &mocks.Configuration{
+				CtrlClientObjects:    []runtime.Object{mockFramework, mockJob},
+				DynamicClientObjects: []runtime.Object{},
+			},
+			options: jobExportOptions{
+				name: "hello",
+			},
+			wantYamlData: `backoffLimit: 6
+completions: 1
+containers:
+- command:
+  - ls
+  - /
+  image: ubuntu
+  name: lister
+description: test
+framework: myframework
+name: hello
+parallelism: 1
+policy:
+  restartPolicy: Never
+type: Job
+version: v1
+`,
+		},
+		{
+			name: "error - no job found",
+			cfg: &mocks.Configuration{
+				CtrlClientObjects:    []runtime.Object{mockFramework},
+				DynamicClientObjects: []runtime.Object{},
+			},
+			options: jobExportOptions{
+				name: "hello",
+			},
+			wantErr: `failed to get job: jobs.theketch.io "hello" not found`,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			out := &bytes.Buffer{}
+			err := jobExport(context.Background(), tt.cfg, tt.options, out)
+			if len(tt.wantErr) > 0 {
+				require.NotNil(t, err)
+				require.Equal(t, tt.wantErr, err.Error())
+				return
+			} else {
+				require.Nil(t, err)
+			}
+			require.Equal(t, tt.wantYamlData, out.String())
+		})
+	}
+}
diff --git a/cmd/ketch/job_list.go b/cmd/ketch/job_list.go
new file mode 100644
index 00000000..5be2483e
--- /dev/null
+++ b/cmd/ketch/job_list.go
@@ -0,0 +1,78 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"strings"
+
+	"github.com/spf13/cobra"
+
+	"github.com/shipa-corp/ketch/cmd/ketch/output"
+	ketchv1 "github.com/shipa-corp/ketch/internal/api/v1beta1"
+)
+
+const jobListHelp = `
+List all jobs.
+`
+
+type jobListOutput struct {
+	Name        string `json:"name"`
+	Version     string `json:"version"`
+	Framework   string `json:"framework"`
+	Description string `json:"description"`
+}
+
+func newJobListCmd(cfg config, out io.Writer) *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "list",
+		Short: "List all jobs.",
+		Long:  jobListHelp,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			return jobList(cmd.Context(), cfg, out)
+		},
+	}
+	return cmd
+}
+
+func jobList(ctx context.Context, cfg config, out io.Writer) error {
+	jobs := ketchv1.JobList{}
+	if err := cfg.Client().List(ctx, &jobs); err != nil {
+		return fmt.Errorf("failed to get list of jobs: %w", err)
+	}
+	return output.Write(generateJobListOutput(jobs), out, "column")
+}
+
+func generateJobListOutput(jobs ketchv1.JobList) []jobListOutput {
+	var output []jobListOutput
+	for _, item := range jobs.Items {
+		output = append(output, jobListOutput{
+			Name:        item.Name,
+			Version:     item.Spec.Version,
+			Framework:   item.Spec.Framework,
+			Description: item.Spec.Description,
+		})
+	}
+	return output
+}
+
+func jobListNames(cfg config, nameFilter ...string) ([]string, error) {
+	jobs := ketchv1.JobList{}
+	if err := cfg.Client().List(context.TODO(), &jobs); err != nil {
+		return nil, fmt.Errorf("failed to list apps: %w", err)
+	}
+
+	jobNames := make([]string, 0)
+	for _, j := range jobs.Items {
+		if len(nameFilter) == 0 {
+			jobNames = append(jobNames, j.Name)
+		}
+
+		for _, filter := range nameFilter {
+			if strings.Contains(j.Name, filter) {
+				jobNames = append(jobNames, j.Name)
+			}
+		}
+	}
+	return jobNames, nil
+}
diff --git a/cmd/ketch/job_list_test.go b/cmd/ketch/job_list_test.go
new file mode 100644
index 00000000..6b7f00ad
--- /dev/null
+++ b/cmd/ketch/job_list_test.go
@@ -0,0 +1,126 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+
+	ketchv1 "github.com/shipa-corp/ketch/internal/api/v1beta1"
+	"github.com/shipa-corp/ketch/internal/mocks"
+)
+
+func TestJobList(t *testing.T) {
+	mockJob := &ketchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{Name: "hello", Namespace: "ketch-myframework"},
+		Spec: ketchv1.JobSpec{
+			Name:         "hello",
+			Version:      "v1",
+			Framework:    "myframework",
+			Description:  "test",
+			Parallelism:  1,
+			Completions:  1,
+			Suspend:      false,
+			BackoffLimit: 6,
+			Containers: []ketchv1.Container{
+				{
+					Name:    "lister",
+					Image:   "ubuntu",
+					Command: []string{"ls", "/"},
+				},
+			},
+			Policy: ketchv1.Policy{
+				RestartPolicy: "Never",
+			},
+			Type: "Job",
+		},
+	}
+	tests := []struct {
+		name    string
+		cfg     config
+		wantOut string
+		wantErr string
+	}{
+		{
+			name: "success",
+			cfg: &mocks.Configuration{
+				CtrlClientObjects:    []runtime.Object{mockJob},
+				DynamicClientObjects: []runtime.Object{},
+			},
+			wantOut: "NAME     VERSION    FRAMEWORK      DESCRIPTION\nhello    v1         myframework    test\n",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			out := &bytes.Buffer{}
+			err := jobList(context.Background(), tt.cfg, out)
+			if len(tt.wantErr) > 0 {
+				require.NotNil(t, err)
+				require.Equal(t, tt.wantErr, err.Error())
+				return
+			} else {
+				require.Nil(t, err)
+			}
+			require.Equal(t, tt.wantOut, out.String())
+		})
+	}
+}
+
+func TestJobListNames(t *testing.T) {
+	mockJob := &ketchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{Name: "hello", Namespace: "ketch-myframework"},
+		Spec: ketchv1.JobSpec{
+			Name:         "hello",
+			Version:      "v1",
+			Framework:    "myframework",
+			Description:  "test",
+			Parallelism:  1,
+			Completions:  1,
+			Suspend:      false,
+			BackoffLimit: 6,
+			Containers: []ketchv1.Container{
+				{
+					Name:    "lister",
+					Image:   "ubuntu",
+					Command: []string{"ls", "/"},
+				},
+			},
+			Policy: ketchv1.Policy{
+				RestartPolicy: "Never",
+			},
+			Type: "Job",
+		},
+	}
+	tests := []struct {
+		name    string
+		cfg     config
+		filter  string
+		wantOut []string
+		wantErr string
+	}{
+		{
+			name: "success",
+			cfg: &mocks.Configuration{
+				CtrlClientObjects:    []runtime.Object{mockJob},
+				DynamicClientObjects: []runtime.Object{},
+			},
+			wantOut: []string{"hello"},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			out, err := jobListNames(tt.cfg, tt.filter)
+			if len(tt.wantErr) > 0 {
+				require.NotNil(t, err)
+				require.Equal(t, tt.wantErr, err.Error())
+				return
+			} else {
+				require.Nil(t, err)
+			}
+			require.Equal(t, tt.wantOut, out)
+		})
+	}
+}
diff --git a/cmd/ketch/job_remove.go b/cmd/ketch/job_remove.go
new file mode 100644
index 00000000..409b436d
--- /dev/null
+++ b/cmd/ketch/job_remove.go
@@ -0,0 +1,48 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"io"
+
+	"github.com/shipa-corp/ketch/internal/validation"
+
+	"github.com/spf13/cobra"
+)
+
+const jobRemoveHelp = `
+Remove a job.
+`
+
+func newJobRemoveCmd(cfg config, out io.Writer) *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "remove [NAME]",
+		Short: "Remove a job.",
+		Long:  jobRemoveHelp,
+		Args:  cobra.ExactValidArgs(1),
+		RunE: func(cmd *cobra.Command, args []string) error {
+			jobName := args[0]
+			if !validation.ValidateName(jobName) {
+				return ErrInvalidJobName
+			}
+			return jobRemove(cmd.Context(), cfg, jobName, out)
+		},
+		ValidArgsFunction: func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
+			return autoCompleteJobNames(cfg, toComplete)
+		},
+	}
+	return cmd
+}
+
+func jobRemove(ctx context.Context, cfg config, jobName string, out io.Writer) error {
+	job, err := getJob(ctx, cfg, jobName)
+	if err != nil {
+		return err
+	}
+
+	if err := cfg.Client().Delete(ctx, job); err != nil {
+		return fmt.Errorf("failed to delete job: %w", err)
+	}
+	fmt.Fprintln(out, "Successfully removed!")
+	return nil
+}
diff --git a/cmd/ketch/job_remove_test.go b/cmd/ketch/job_remove_test.go
new file mode 100644
index 00000000..ec9a40bd
--- /dev/null
+++ b/cmd/ketch/job_remove_test.go
@@ -0,0 +1,94 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+
+	ketchv1 "github.com/shipa-corp/ketch/internal/api/v1beta1"
+	"github.com/shipa-corp/ketch/internal/mocks"
+	"github.com/shipa-corp/ketch/internal/utils/conversions"
+)
+
+func TestJobRemove(t *testing.T) {
+	mockFramework := &ketchv1.Framework{ObjectMeta: metav1.ObjectMeta{Name: "myframework"}, Spec: ketchv1.FrameworkSpec{
+		Version:       "v1",
+		NamespaceName: "ketch-myframework",
+		Name:          "myframework",
+		AppQuotaLimit: conversions.IntPtr(1),
+		IngressController: ketchv1.IngressControllerSpec{
+			ClassName:       "traefik",
+			ServiceEndpoint: "10.10.20.30",
+			IngressType:     "traefik",
+			ClusterIssuer:   "letsencrypt",
+		},
+	}, Status: ketchv1.FrameworkStatus{Jobs: []string{"hello"}}}
+	mockJob := &ketchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{Name: "hello", Namespace: mockFramework.Spec.NamespaceName},
+		Spec: ketchv1.JobSpec{
+			Name:         "hello",
+			Version:      "v1",
+			Framework:    "myframework",
+			Description:  "test",
+			Parallelism:  1,
+			Completions:  1,
+			Suspend:      false,
+			BackoffLimit: 6,
+			Containers: []ketchv1.Container{
+				{
+					Name:    "lister",
+					Image:   "ubuntu",
+					Command: []string{"ls", "/"},
+				},
+			},
+			Policy: ketchv1.Policy{
+				RestartPolicy: "Never",
+			},
+			Type: "Job",
+		},
+	}
+	tests := []struct {
+		name    string
+		jobName string
+		cfg     config
+		wantOut string
+		wantErr string
+	}{
+		{
+			name:    "success",
+			jobName: mockJob.Name,
+			cfg: &mocks.Configuration{
+				CtrlClientObjects:    []runtime.Object{mockJob, mockFramework},
+				DynamicClientObjects: []runtime.Object{},
+			},
+			wantOut: "Successfully removed!\n",
+		},
+		{
+			name:    "error - job not found",
+			jobName: mockJob.Name,
+			cfg: &mocks.Configuration{
+				CtrlClientObjects:    []runtime.Object{mockFramework},
+				DynamicClientObjects: []runtime.Object{},
+			},
+			wantErr: `failed to get job: jobs.theketch.io "hello" not found`,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			out := &bytes.Buffer{}
+			err := jobRemove(context.Background(), tt.cfg, tt.jobName, out)
+			if len(tt.wantErr) > 0 {
+				require.NotNil(t, err)
+				require.Equal(t, tt.wantErr, err.Error())
+				return
+			} else {
+				require.Nil(t, err)
+			}
+			require.Equal(t, tt.wantOut, out.String())
+		})
+	}
+}
diff --git a/cmd/ketch/root.go b/cmd/ketch/root.go
index b2c3c412..58015a6b 100644
--- a/cmd/ketch/root.go
+++ b/cmd/ketch/root.go
@@ -62,6 +62,7 @@ func newRootCmd(cfg config, out io.Writer, packSvc *pack.Client, ketchConfig con
 	cmd.AddCommand(newCnameCmd(cfg, out))
 	cmd.AddCommand(newFrameworkCmd(cfg, out))
 	cmd.AddCommand(newEnvCmd(cfg, out))
+	cmd.AddCommand(newJobCmd(cfg, out))
 	cmd.AddCommand(newCompletionCmd())
 	return cmd
 }
diff --git a/config/crd/bases/theketch.io_jobs.yaml b/config/crd/bases/theketch.io_jobs.yaml
index 4f440df6..2e3d21a5 100644
--- a/config/crd/bases/theketch.io_jobs.yaml
+++ b/config/crd/bases/theketch.io_jobs.yaml
@@ -71,8 +71,6 @@ spec:
               properties:
                 restartPolicy:
                   type: string
-              required:
-              - restartPolicy
               type: object
             suspend:
               type: boolean
@@ -81,12 +79,9 @@ spec:
             version:
               type: string
           required:
-          - description
           - framework
           - name
-          - policy
           - type
-          - version
           type: object
         status:
           description: JobStatus defines the observed state of Job
diff --git a/internal/api/v1beta1/job_types.go b/internal/api/v1beta1/job_types.go
index 7d58a5a0..02c6f458 100644
--- a/internal/api/v1beta1/job_types.go
+++ b/internal/api/v1beta1/job_types.go
@@ -1,6 +1,5 @@
 /*
 Copyright 2021.
-
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
 You may obtain a copy of the License at
@@ -18,23 +17,24 @@ package v1beta1
 
 import (
 	"github.com/shipa-corp/ketch/internal/templates"
+
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 // JobSpec defines the desired state of Job
 type JobSpec struct {
-	Version      string      `json:"version"`
+	Version      string      `json:"version,omitempty"`
 	Type         string      `json:"type"`
 	Name         string      `json:"name"`
 	Framework    string      `json:"framework"`
-	Description  string      `json:"description"`
+	Description  string      `json:"description,omitempty"`
 	Parallelism  int         `json:"parallelism,omitempty"`
 	Completions  int         `json:"completions,omitempty"`
 	Suspend      bool        `json:"suspend,omitempty"`
 	BackoffLimit int         `json:"backoffLimit,omitempty"`
 	Containers   []Container `json:"containers,omitempty"`
-	Policy       Policy      `json:"policy"`
+	Policy       Policy      `json:"policy,omitempty"`
 }
 
 // JobStatus defines the observed state of Job
@@ -66,7 +66,7 @@ type JobList struct {
 
 // Policy represents the policy types a job can have
 type Policy struct {
-	RestartPolicy RestartPolicy `json:"restartPolicy"`
+	RestartPolicy RestartPolicy `json:"restartPolicy,omitempty"`
 }
 
 // Container represents a single container run in a Job
diff --git a/internal/templates/job/yamls/job.yaml b/internal/templates/job/yamls/job.yaml
index f48afd97..749a5bef 100644
--- a/internal/templates/job/yamls/job.yaml
+++ b/internal/templates/job/yamls/job.yaml
@@ -7,16 +7,16 @@ metadata:
   name: {{ $.Values.job.name }}
 spec:
   {{- if $.Values.job.parallelism }}
-{{ $.Values.job.parallelism | toYaml | indent 4 }}
+  parallelism: {{ $.Values.job.parallelism }}
   {{- end }}
   {{- if $.Values.job.parallelism }}
-{{ $.Values.job.parallelism | toYaml | indent 4 }}
+  completions: {{ $.Values.job.completions }}
   {{- end }}
   {{- if $.Values.job.backoffLimit }}
-{{ $.Values.job.backoffLimit | toYaml | indent 4 }}
+  backoffLimit: {{ $.Values.job.backoffLimit }}
   {{- end }}
   {{- if $.Values.job.suspend }}
-{{ $.Values.job.suspend | toYaml | indent 4 }}
+  suspend: {{ $.Values.job.suspend }}
   {{- end }}
   template:
     spec: