def METHOD_NAME(client, jobflow_id, job_name, jar_path, main_class, extra_args):
  """Submits single spark job to a running cluster"""
  spark_job = {
    'Name': job_name,
    'ActionOnFailure': 'CONTINUE',
    'HadoopJarStep': {
      'Jar': 'command-runner.jar'
    }
  }
  spark_args = ['spark-submit', "--deploy-mode", "cluster"]
  if main_class:
    spark_args.extend(['--class', main_class])
  spark_args.extend([jar_path])
  spark_args.extend(extra_args)
  spark_job['HadoopJarStep']['Args'] = spark_args
  try:
    response = client.add_job_flow_steps(
            JobFlowId=jobflow_id,
            Steps=[spark_job],
    )
  except ClientError as e:
    print(e.response['Error']['Message'])
    exit(1)
  step_id = response['StepIds'][0]
  print("Step Id {} has been submitted".format(step_id))
  return step_id