def METHOD_NAME(benchmark_spec):
  """Executes the given jar on the specified Spark cluster.
  Args:
    benchmark_spec: The benchmark specification. Contains all data that is
        required to run the benchmark.
  Returns:
    A list of sample.Sample objects.
  """
  spark_cluster = benchmark_spec.spark_service
  start = datetime.datetime.now()
  terasort_jar = spark_cluster.GetExampleJar(spark_service.HADOOP_JOB_TYPE)
  results = []
  metadata = copy.copy(spark_cluster.GetMetadata())
  logging.info('metadata %s ' % str(metadata))
  base_dir = FLAGS.terasort_data_base
  if FLAGS.terasort_append_timestamp:
    time_string = datetime.datetime.now().strftime('%Y%m%d%H%S')
    base_dir += time_string
    base_dir += '/'
  unsorted_dir = base_dir + 'unsorted'
  sorted_dir = base_dir + 'sorted'
  validate_dir = base_dir + 'validate'
  metadata.update({'terasort_num_rows': FLAGS.terasort_num_rows,
                   'terasort_sorted_dir': sorted_dir,
                   'terasort_unsorted_dir': unsorted_dir,
                   'terasort_validate_dir': validate_dir})
  gen_args = [TERAGEN, str(FLAGS.terasort_num_rows), unsorted_dir]
  sort_args = [TERASORT, unsorted_dir, sorted_dir]
  validate_args = [TERAVALIDATE, sorted_dir, validate_dir]
  stages = [('generate', gen_args),
            ('sort', sort_args),
            ('validate', validate_args)]
  cumulative_runtime = 0
  for (label, args) in stages:
    stats = spark_cluster.SubmitJob(terasort_jar,
                                    None,
                                    job_type=spark_service.HADOOP_JOB_TYPE,
                                    job_arguments=args)
    if not stats[spark_service.SUCCESS]:
      raise Exception('Stage {0} unsuccessful'.format(label))
    current_time = datetime.datetime.now()
    wall_time = (current_time - start).total_seconds()
    results.append(sample.Sample(label + '_wall_time',
                                 wall_time,
                                 'seconds', metadata))
    start = current_time
    if spark_service.RUNTIME in stats:
      results.append(sample.Sample(label + '_runtime',
                                   stats[spark_service.RUNTIME],
                                   'seconds', metadata))
    cumulative_runtime += stats[spark_service.RUNTIME]
    if spark_service.WAITING in stats:
      results.append(sample.Sample(label + '_pending_time',
                                   stats[spark_service.WAITING],
                                   'seconds', metadata))
  results.append(sample.Sample('cumulative_runtime',
                               cumulative_runtime,
                               'seconds', metadata))
  if not spark_cluster.user_managed:
    create_time = (spark_cluster.resource_ready_time -
                   spark_cluster.create_start_time)
    results.append(sample.Sample('cluster_create_time',
                                 create_time,
                                 'seconds', metadata))
  return results