diff --git a/pkg/kubectl/cmd/create/BUILD b/pkg/kubectl/cmd/create/BUILD
index fedfef695f..3cf3f5b4be 100644
--- a/pkg/kubectl/cmd/create/BUILD
+++ b/pkg/kubectl/cmd/create/BUILD
@@ -94,10 +94,8 @@ go_test(
         "//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/diff:go_default_library",
-        "//staging/src/k8s.io/client-go/kubernetes/fake:go_default_library",
         "//staging/src/k8s.io/client-go/rest:go_default_library",
         "//staging/src/k8s.io/client-go/rest/fake:go_default_library",
-        "//staging/src/k8s.io/client-go/testing:go_default_library",
         "//vendor/github.com/stretchr/testify/assert:go_default_library",
     ],
 )
diff --git a/pkg/kubectl/cmd/create/create_job.go b/pkg/kubectl/cmd/create/create_job.go
index 1e92754def..62385d9e25 100644
--- a/pkg/kubectl/cmd/create/create_job.go
+++ b/pkg/kubectl/cmd/create/create_job.go
@@ -23,9 +23,10 @@ import (
 
 	batchv1 "k8s.io/api/batch/v1"
 	batchv1beta1 "k8s.io/api/batch/v1beta1"
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
-	clientbatchv1 "k8s.io/client-go/kubernetes/typed/batch/v1"
+	batchv1client "k8s.io/client-go/kubernetes/typed/batch/v1"
 	"k8s.io/kubernetes/pkg/api/legacyscheme"
 	"k8s.io/kubernetes/pkg/kubectl/cmd/templates"
 	cmdutil "k8s.io/kubernetes/pkg/kubectl/cmd/util"
@@ -40,6 +41,12 @@ var (
 		Create a job with the specified name.`))
 
 	jobExample = templates.Examples(i18n.T(`
+		# Create a job
+		kubectl create job my-job --image=busybox
+
+		# Create a job with command
+		kubectl create job my-job --image=busybox -- date
+
 		# Create a job from a CronJob named "a-cronjob"
 		kubectl create job test-job --from=cronjob/a-cronjob`))
 )
@@ -49,15 +56,16 @@ type CreateJobOptions struct {
 
 	PrintObj func(obj runtime.Object) error
 
-	Name string
-	From string
+	Name    string
+	Image   string
+	From    string
+	Command []string
 
-	Namespace    string
-	OutputFormat string
-	Client       clientbatchv1.BatchV1Interface
-	DryRun       bool
-	Builder      *resource.Builder
-	Cmd          *cobra.Command
+	Namespace string
+	Client    batchv1client.BatchV1Interface
+	DryRun    bool
+	Builder   *resource.Builder
+	Cmd       *cobra.Command
 
 	genericclioptions.IOStreams
 }
@@ -72,15 +80,16 @@ func NewCreateJobOptions(ioStreams genericclioptions.IOStreams) *CreateJobOption
 // NewCmdCreateJob is a command to ease creating Jobs from CronJobs.
 func NewCmdCreateJob(f cmdutil.Factory, ioStreams genericclioptions.IOStreams) *cobra.Command {
 	o := NewCreateJobOptions(ioStreams)
-
 	cmd := &cobra.Command{
-		Use:     "job NAME [--from=CRONJOB]",
+		Use:     "job NAME [--image=image --from=cronjob/name] -- [COMMAND] [args...]",
 		Short:   jobLong,
 		Long:    jobLong,
 		Example: jobExample,
 		Run: func(cmd *cobra.Command, args []string) {
-			cmdutil.CheckErr(o.Complete(f, cmd, args))
-			cmdutil.CheckErr(o.RunCreateJob())
+			argsLenAtDash := cmd.ArgsLenAtDash()
+			cmdutil.CheckErr(o.Complete(f, cmd, args, argsLenAtDash))
+			cmdutil.CheckErr(o.Validate())
+			cmdutil.CheckErr(o.Run())
 		},
 	}
 
@@ -89,32 +98,38 @@ func NewCmdCreateJob(f cmdutil.Factory, ioStreams genericclioptions.IOStreams) *
 	cmdutil.AddApplyAnnotationFlags(cmd)
 	cmdutil.AddValidateFlags(cmd)
 	cmdutil.AddDryRunFlag(cmd)
+	cmd.Flags().StringVar(&o.Image, "image", o.Image, "Image name to run.")
 	cmd.Flags().StringVar(&o.From, "from", o.From, "The name of the resource to create a Job from (only cronjob is supported).")
 
 	return cmd
 }
 
-func (o *CreateJobOptions) Complete(f cmdutil.Factory, cmd *cobra.Command, args []string) (err error) {
-	if len(args) == 0 {
+func (o *CreateJobOptions) Complete(f cmdutil.Factory, cmd *cobra.Command, args []string, argsLenAtDash int) error {
+	if len(args) == 0 || argsLenAtDash == 0 {
 		return cmdutil.UsageErrorf(cmd, "NAME is required")
 	}
 	o.Name = args[0]
+	if len(args) > 1 {
+		o.Command = args[1:]
+	}
+
+	clientConfig, err := f.ToRESTConfig()
+	if err != nil {
+		return err
+	}
+	o.Client, err = batchv1client.NewForConfig(clientConfig)
+	if err != nil {
+		return err
+	}
 
 	o.Namespace, _, err = f.ToRawKubeConfigLoader().Namespace()
 	if err != nil {
 		return err
 	}
-
-	clientset, err := f.KubernetesClientSet()
-	if err != nil {
-		return err
-	}
-	o.Client = clientset.BatchV1()
 	o.Builder = f.NewBuilder()
-	o.DryRun = cmdutil.GetDryRunFlag(cmd)
 	o.Cmd = cmd
-	o.OutputFormat = cmdutil.GetFlagString(cmd, "output")
 
+	o.DryRun = cmdutil.GetDryRunFlag(cmd)
 	if o.DryRun {
 		o.PrintFlags.Complete("%s (dry run)")
 	}
@@ -122,7 +137,6 @@ func (o *CreateJobOptions) Complete(f cmdutil.Factory, cmd *cobra.Command, args
 	if err != nil {
 		return err
 	}
-
 	o.PrintObj = func(obj runtime.Object) error {
 		return printer.PrintObj(obj, o.Out)
 	}
@@ -130,50 +144,47 @@ func (o *CreateJobOptions) Complete(f cmdutil.Factory, cmd *cobra.Command, args
 	return nil
 }
 
-func (o *CreateJobOptions) RunCreateJob() error {
-	infos, err := o.Builder.
-		Unstructured().
-		NamespaceParam(o.Namespace).DefaultNamespace().
-		ResourceTypeOrNameArgs(false, o.From).
-		Flatten().
-		Latest().
-		Do().
-		Infos()
-	if err != nil {
-		return err
+func (o *CreateJobOptions) Validate() error {
+	if (len(o.Image) == 0 && len(o.From) == 0) || (len(o.Image) != 0 && len(o.From) != 0) {
+		return fmt.Errorf("either --image or --from must be specified")
 	}
-	if len(infos) != 1 {
-		return fmt.Errorf("from must be an existing cronjob")
+	if o.Command != nil && len(o.Command) != 0 && len(o.From) != 0 {
+		return fmt.Errorf("cannot specify --from and command")
 	}
-
-	uncastVersionedObj, err := scheme.Scheme.ConvertToVersion(infos[0].Object, batchv1beta1.SchemeGroupVersion)
-	if err != nil {
-		return fmt.Errorf("from must be an existing cronjob: %v", err)
-	}
-	cronJob, ok := uncastVersionedObj.(*batchv1beta1.CronJob)
-	if !ok {
-		return fmt.Errorf("from must be an existing cronjob")
-	}
-
-	return o.createJob(cronJob)
+	return nil
 }
 
-func (o *CreateJobOptions) createJob(cronJob *batchv1beta1.CronJob) error {
-	annotations := make(map[string]string)
-	annotations["cronjob.kubernetes.io/instantiate"] = "manual"
-	for k, v := range cronJob.Spec.JobTemplate.Annotations {
-		annotations[k] = v
-	}
-	job := &batchv1.Job{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:        o.Name,
-			Namespace:   o.Namespace,
-			Annotations: annotations,
-			Labels:      cronJob.Spec.JobTemplate.Labels,
-		},
-		Spec: cronJob.Spec.JobTemplate.Spec,
-	}
+func (o *CreateJobOptions) Run() error {
+	var job *batchv1.Job
+	if len(o.Image) > 0 {
+		job = o.createJob()
+	} else {
+		infos, err := o.Builder.
+			Unstructured().
+			NamespaceParam(o.Namespace).DefaultNamespace().
+			ResourceTypeOrNameArgs(false, o.From).
+			Flatten().
+			Latest().
+			Do().
+			Infos()
+		if err != nil {
+			return err
+		}
+		if len(infos) != 1 {
+			return fmt.Errorf("from must be an existing cronjob")
+		}
 
+		uncastVersionedObj, err := scheme.Scheme.ConvertToVersion(infos[0].Object, batchv1beta1.SchemeGroupVersion)
+		if err != nil {
+			return fmt.Errorf("from must be an existing cronjob: %v", err)
+		}
+		cronJob, ok := uncastVersionedObj.(*batchv1beta1.CronJob)
+		if !ok {
+			return fmt.Errorf("from must be an existing cronjob")
+		}
+
+		job = o.createJobFromCronJob(cronJob)
+	}
 	if !o.DryRun {
 		var err error
 		job, err = o.Client.Jobs(o.Namespace).Create(job)
@@ -184,3 +195,45 @@ func (o *CreateJobOptions) createJob(cronJob *batchv1beta1.CronJob) error {
 
 	return o.PrintObj(job)
 }
+
+func (o *CreateJobOptions) createJob() *batchv1.Job {
+	return &batchv1.Job{
+		// this is ok because we know exactly how we want to be serialized
+		TypeMeta: metav1.TypeMeta{APIVersion: batchv1.SchemeGroupVersion.String(), Kind: "Job"},
+		ObjectMeta: metav1.ObjectMeta{
+			Name: o.Name,
+		},
+		Spec: batchv1.JobSpec{
+			Template: corev1.PodTemplateSpec{
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name:    o.Name,
+							Image:   o.Image,
+							Command: o.Command,
+						},
+					},
+					RestartPolicy: corev1.RestartPolicyNever,
+				},
+			},
+		},
+	}
+}
+
+func (o *CreateJobOptions) createJobFromCronJob(cronJob *batchv1beta1.CronJob) *batchv1.Job {
+	annotations := make(map[string]string)
+	annotations["cronjob.kubernetes.io/instantiate"] = "manual"
+	for k, v := range cronJob.Spec.JobTemplate.Annotations {
+		annotations[k] = v
+	}
+	return &batchv1.Job{
+		// this is ok because we know exactly how we want to be serialized
+		TypeMeta: metav1.TypeMeta{APIVersion: batchv1.SchemeGroupVersion.String(), Kind: "Job"},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:        o.Name,
+			Annotations: annotations,
+			Labels:      cronJob.Spec.JobTemplate.Labels,
+		},
+		Spec: cronJob.Spec.JobTemplate.Spec,
+	}
+}
diff --git a/pkg/kubectl/cmd/create/create_job_test.go b/pkg/kubectl/cmd/create/create_job_test.go
index 7fa4b9e78c..317010c287 100644
--- a/pkg/kubectl/cmd/create/create_job_test.go
+++ b/pkg/kubectl/cmd/create/create_job_test.go
@@ -17,121 +17,176 @@ limitations under the License.
 package create
 
 import (
+	"strings"
 	"testing"
 
 	batchv1 "k8s.io/api/batch/v1"
 	batchv1beta1 "k8s.io/api/batch/v1beta1"
 	corev1 "k8s.io/api/core/v1"
+	apiequality "k8s.io/apimachinery/pkg/api/equality"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/runtime"
-	fake "k8s.io/client-go/kubernetes/fake"
-	clienttesting "k8s.io/client-go/testing"
-	"k8s.io/kubernetes/pkg/api/legacyscheme"
-	cmdtesting "k8s.io/kubernetes/pkg/kubectl/cmd/testing"
-	"k8s.io/kubernetes/pkg/kubectl/genericclioptions"
 )
 
-func TestCreateJobFromCronJob(t *testing.T) {
-	var submittedJob *batchv1.Job
-	testNamespaceName := "test"
-	testCronJobName := "test-cronjob"
-	testJobName := "test-job"
-	testImageName := "fake"
-
-	expectedLabels := make(map[string]string)
-	expectedAnnotations := make(map[string]string)
-	expectedLabels["test-label"] = "test-value"
-
-	expectJob := &batchv1.Job{
-		ObjectMeta: metav1.ObjectMeta{
-			Namespace:   testNamespaceName,
-			Labels:      expectedLabels,
-			Annotations: expectedAnnotations,
+func TestCreateJobValidation(t *testing.T) {
+	tests := map[string]struct {
+		image    string
+		command  []string
+		from     string
+		expected string
+	}{
+		"empty flags": {
+			expected: "--image or --from must be specified",
 		},
-		Spec: batchv1.JobSpec{
-			Template: corev1.PodTemplateSpec{
-				Spec: corev1.PodSpec{
-					Containers: []corev1.Container{
-						{Image: testImageName},
+		"both image and from specified": {
+			image:    "my-image",
+			from:     "cronjob/xyz",
+			expected: "--image or --from must be specified",
+		},
+		"from and command specified": {
+			from:     "cronjob/xyz",
+			command:  []string{"test", "command"},
+			expected: "cannot specify --from and command",
+		},
+	}
+
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			o := &CreateJobOptions{
+				Image:   tc.image,
+				From:    tc.from,
+				Command: tc.command,
+			}
+
+			err := o.Validate()
+			if err != nil && !strings.Contains(err.Error(), tc.expected) {
+				t.Errorf("unexpected error: %v", err)
+			}
+		})
+	}
+}
+
+func TestCreateJob(t *testing.T) {
+	jobName := "test-job"
+	tests := map[string]struct {
+		image    string
+		command  []string
+		expected *batchv1.Job
+	}{
+		"just image": {
+			image: "busybox",
+			expected: &batchv1.Job{
+				TypeMeta: metav1.TypeMeta{APIVersion: batchv1.SchemeGroupVersion.String(), Kind: "Job"},
+				ObjectMeta: metav1.ObjectMeta{
+					Name: jobName,
+				},
+				Spec: batchv1.JobSpec{
+					Template: corev1.PodTemplateSpec{
+						Spec: corev1.PodSpec{
+							Containers: []corev1.Container{
+								{
+									Name:  jobName,
+									Image: "busybox",
+								},
+							},
+							RestartPolicy: corev1.RestartPolicyNever,
+						},
+					},
+				},
+			},
+		},
+		"image and command": {
+			image:   "busybox",
+			command: []string{"date"},
+			expected: &batchv1.Job{
+				TypeMeta: metav1.TypeMeta{APIVersion: batchv1.SchemeGroupVersion.String(), Kind: "Job"},
+				ObjectMeta: metav1.ObjectMeta{
+					Name: jobName,
+				},
+				Spec: batchv1.JobSpec{
+					Template: corev1.PodTemplateSpec{
+						Spec: corev1.PodSpec{
+							Containers: []corev1.Container{
+								{
+									Name:    jobName,
+									Image:   "busybox",
+									Command: []string{"date"},
+								},
+							},
+							RestartPolicy: corev1.RestartPolicyNever,
+						},
 					},
 				},
 			},
 		},
 	}
 
-	cronJob := &batchv1beta1.CronJob{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      testCronJobName,
-			Namespace: testNamespaceName,
-		},
-		Spec: batchv1beta1.CronJobSpec{
-			Schedule: "* * * * *",
-			JobTemplate: batchv1beta1.JobTemplateSpec{
-				ObjectMeta: metav1.ObjectMeta{
-					Namespace: testNamespaceName,
-					Labels:    expectedLabels,
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			o := &CreateJobOptions{
+				Name:    jobName,
+				Image:   tc.image,
+				Command: tc.command,
+			}
+			job := o.createJob()
+			if !apiequality.Semantic.DeepEqual(job, tc.expected) {
+				t.Errorf("expected:\n%#v\ngot:\n%#v", tc.expected, job)
+			}
+		})
+	}
+}
+
+func TestCreateJobFromCronJob(t *testing.T) {
+	jobName := "test-job"
+	tests := map[string]struct {
+		from     *batchv1beta1.CronJob
+		expected *batchv1.Job
+	}{
+		"from CronJob": {
+			from: &batchv1beta1.CronJob{
+				Spec: batchv1beta1.CronJobSpec{
+					JobTemplate: batchv1beta1.JobTemplateSpec{
+						Spec: batchv1.JobSpec{
+							Template: corev1.PodTemplateSpec{
+								Spec: corev1.PodSpec{
+									Containers: []corev1.Container{
+										{Image: "test-image"},
+									},
+									RestartPolicy: corev1.RestartPolicyNever,
+								},
+							},
+						},
+					},
+				},
+			},
+			expected: &batchv1.Job{
+				TypeMeta: metav1.TypeMeta{APIVersion: batchv1.SchemeGroupVersion.String(), Kind: "Job"},
+				ObjectMeta: metav1.ObjectMeta{
+					Name:        jobName,
+					Annotations: map[string]string{"cronjob.kubernetes.io/instantiate": "manual"},
+				},
+				Spec: batchv1.JobSpec{
+					Template: corev1.PodTemplateSpec{
+						Spec: corev1.PodSpec{
+							Containers: []corev1.Container{
+								{Image: "test-image"},
+							},
+							RestartPolicy: corev1.RestartPolicyNever,
+						},
+					},
 				},
-				Spec: expectJob.Spec,
 			},
 		},
 	}
 
-	clientset := fake.Clientset{}
-	clientset.PrependReactor("create", "jobs", func(action clienttesting.Action) (handled bool, ret runtime.Object, err error) {
-		ca := action.(clienttesting.CreateAction)
-		submittedJob = ca.GetObject().(*batchv1.Job)
-		return true, expectJob, nil
-	})
-	f := cmdtesting.NewTestFactory()
-	defer f.Cleanup()
-
-	printFlags := genericclioptions.NewPrintFlags("created").WithTypeSetter(legacyscheme.Scheme)
-
-	ioStreams, _, buf, _ := genericclioptions.NewTestIOStreams()
-	cmdOptions := &CreateJobOptions{
-		PrintFlags: printFlags,
-		Name:       testJobName,
-		Namespace:  testNamespaceName,
-		Client:     clientset.BatchV1(),
-		Cmd:        NewCmdCreateJob(f, ioStreams),
-		PrintObj: func(obj runtime.Object) error {
-			p, err := printFlags.ToPrinter()
-			if err != nil {
-				return err
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			o := &CreateJobOptions{
+				Name: jobName,
 			}
-
-			return p.PrintObj(obj, buf)
-		},
-		IOStreams: ioStreams,
-	}
-
-	err := cmdOptions.createJob(cronJob)
-	if err != nil {
-		t.Errorf("unexpected error: %v", err)
-	}
-
-	if submittedJob.ObjectMeta.Name != testJobName {
-		t.Errorf("expected '%s', got '%s'", testJobName, submittedJob.ObjectMeta.Name)
-	}
-
-	if l := len(submittedJob.Annotations); l != 1 {
-		t.Errorf("expected length of annotations array to be 1, got %d", l)
-	}
-	if v, ok := submittedJob.Annotations["cronjob.kubernetes.io/instantiate"]; !ok || v != "manual" {
-		t.Errorf("expected annotation cronjob.kubernetes.io/instantiate=manual to exist, got '%s'", v)
-	}
-
-	if l := len(submittedJob.Labels); l != 1 {
-		t.Errorf("expected length of labels array to be 1, got %d", l)
-	}
-	if v, ok := submittedJob.Labels["test-label"]; !ok || v != "test-value" {
-		t.Errorf("expected label test-label=test-value to exist, got '%s'", v)
-	}
-
-	if l := len(submittedJob.Spec.Template.Spec.Containers); l != 1 {
-		t.Errorf("expected length of container array to be 1, got %d", l)
-	}
-	if submittedJob.Spec.Template.Spec.Containers[0].Image != testImageName {
-		t.Errorf("expected '%s', got '%s'", testImageName, submittedJob.Spec.Template.Spec.Containers[0].Image)
+			job := o.createJobFromCronJob(tc.from)
+			if !apiequality.Semantic.DeepEqual(job, tc.expected) {
+				t.Errorf("expected:\n%#v\ngot:\n%#v", tc.expected, job)
+			}
+		})
 	}
 }
diff --git a/test/cmd/create.sh b/test/cmd/create.sh
index 6221d17b84..5792883f56 100755
--- a/test/cmd/create.sh
+++ b/test/cmd/create.sh
@@ -72,3 +72,39 @@ run_kubectl_create_error_tests() {
   set +o nounset
   set +o errexit
 }
+
+# Runs kubectl create job tests
+run_create_job_tests() {
+    set -o nounset
+    set -o errexit
+
+    create_and_use_new_namespace
+
+    # Test kubectl create job
+    kubectl create job test-job --image=k8s.gcr.io/nginx:test-cmd
+    # Post-Condition: job nginx is created
+    kube::test::get_object_assert 'job test-job' "{{$image_field0}}" 'k8s.gcr.io/nginx:test-cmd'
+    # Clean up
+    kubectl delete job test-job "${kube_flags[@]}"
+
+    # Test kubectl create job with command
+    kubectl create job test-job-pi "--image=$IMAGE_PERL" -- perl -Mbignum=bpi -wle 'print bpi(20)'
+    kube::test::get_object_assert 'job test-job-pi' "{{$image_field0}}" $IMAGE_PERL
+    # Clean up
+    kubectl delete job test-job-pi
+
+    # Test kubectl create job from cronjob
+    # Pre-Condition: create a cronjob
+    kubectl run test-pi --schedule="* */5 * * *" --generator=cronjob/v1beta1 "--image=$IMAGE_PERL" --restart=OnFailure -- perl -Mbignum=bpi -wle 'print bpi(10)'
+    kubectl create job my-pi --from=cronjob/test-pi
+    # Post-condition: container args contain expected command
+    output_message=$(kubectl get job my-pi -o go-template='{{(index .spec.template.spec.containers 0).args}}' "${kube_flags[@]}")
+    kube::test::if_has_string "${output_message}" "perl -Mbignum=bpi -wle print bpi(10)"
+
+    # Clean up
+    kubectl delete job my-pi
+    kubectl delete cronjob test-pi
+
+    set +o nounset
+    set +o errexit
+}
diff --git a/test/cmd/legacy-script.sh b/test/cmd/legacy-script.sh
index 10e781541f..49c849e8b3 100755
--- a/test/cmd/legacy-script.sh
+++ b/test/cmd/legacy-script.sh
@@ -574,6 +574,7 @@ runTests() {
 
   if kube::test::if_supports_resource "${job}" ; then
     record_command run_job_tests
+    record_command run_create_job_tests
   fi
 
   #################
@@ -721,7 +722,7 @@ runTests() {
 
     output_message=$(kubectl auth can-i list jobs.batch/bar -n foo --quiet 2>&1 "${kube_flags[@]}")
     kube::test::if_empty_string "${output_message}"
-    
+
     output_message=$(kubectl auth can-i get pods --subresource=log 2>&1 "${kube_flags[@]}"; echo $?)
     kube::test::if_has_string "${output_message}" '0'