def default_options(self):
     super_opts = super(HadoopRunnerOptionStore, self).default_options()
     return combine_dicts(super_opts, {
         'hadoop_home': os.environ.get('HADOOP_HOME'),
         'hdfs_scratch_dir': 'tmp/mrjob',
         'check_input_paths': True
     })

 def _default_opts(self):
     return combine_dicts(
         super(MRJobBinRunner, self)._default_opts(),
         dict(
             read_logs=True,
         )
     )

 def _default_opts(self):
     return combine_dicts(
         super(MRJobBinRunner, self)._default_opts(),
         dict(
             sh_bin=['sh', '-ex'],
         )
     )

 def _default_opts(self):
     return combine_dicts(
         super(HadoopJobRunner, self)._default_opts(),
         dict(
             hadoop_tmp_dir='tmp/mrjob',
         )
     )

    def jobconf(self):
        orig_jobconf = super(ZipNumClusterJob, self).jobconf()
        custom_jobconf = {'mapreduce.job.reduces': self.options.shards,
                          'mapreduce.totalorderpartitioner.path': self.options.splitfile}

        combined = combine_dicts(orig_jobconf, custom_jobconf)
        return combined

def _add_runner_args_for_opt(parser, opt_name, include_deprecated=True):
    """Add switches for a single option (*opt_name*) to the given parser."""
    conf = _RUNNER_OPTS[opt_name]

    if conf.get('deprecated') and not include_deprecated:
        return

    switches = conf.get('switches') or []

    for args, kwargs in switches:
        kwargs = dict(kwargs)

        deprecated_aliases = kwargs.pop('deprecated_aliases', None)

        kwargs['dest'] = opt_name

        if kwargs.get('action') == 'append':
            kwargs['default'] = []
        else:
            kwargs['default'] = None

        parser.add_argument(*args, **kwargs)

        # add a switch for deprecated aliases
        if deprecated_aliases and include_deprecated:
            help = 'Deprecated alias%s for %s' % (
                ('es' if len(deprecated_aliases) > 1 else ''),
                args[-1])
            parser.add_argument(
                *deprecated_aliases,
                **combine_dicts(kwargs, dict(help=help)))

 def _default_opts(self):
     return combine_dicts(
         super(SparkMRJobRunner, self)._default_opts(),
         dict(
             cloud_part_size_mb=_DEFAULT_CLOUD_PART_SIZE_MB,
         ),
     )

 def _opts_combiners(cls):
     """Map from option name to a combine_*() function used to combine
     values for that option. This allows us to specify that some options
     are lists, or contain environment variables, or whatever."""
     return combine_dicts(
         super(HadoopJobRunner, cls)._opts_combiners(),
         {"hadoop_bin": combine_paths, "hadoop_home": combine_paths, "hdfs_scratch_dir": combine_paths},
     )

    def test_hadoop_1(self):
        updated, warnings = self.updated_and_warnings(
            self.JOBCONF, '1.0')

        self.assertEqual(updated,
                         combine_dicts(self.JOBCONF, {'user.name': 'dave'}))
        self.assertIn('do not match hadoop version', warnings)
        self.assertIn('mapreduce.job.user.name: user.name', warnings)

    def paginate(self, **kwargs):
        result = self.method(**kwargs)

        values = result[self.result_key]

        for page_start in range(0, len(values), self.page_size):
            page = values[page_start:page_start + self.page_size]
            yield combine_dicts(result, {self.result_key: page})

 def _default_opts(self):
     return combine_dicts(
         super(HadoopJobRunner, self)._default_opts(),
         dict(
             hadoop_tmp_dir='tmp/mrjob',
             spark_deploy_mode='client',
             spark_master='yarn',
         )
     )

    def test_hadoop_2(self):
        updated, warnings = self.updated_and_warnings(
            self.JOBCONF, '2.0')

        self.assertEqual(updated,
                         combine_dicts(self.JOBCONF,
                                       {'mapreduce.job.jar': 'a.jar'}))
        self.assertIn('do not match hadoop version', warnings)
        self.assertIn('mapred.jar: mapreduce.job.jar', warnings)

 def _job_runner_kwargs_for_runner(self, runner_alias):
     """Helper method that powers the *_job_runner_kwargs()
     methods."""
     # user can no longer silently ignore switches by overriding
     # job_runner_kwargs()
     return combine_dicts(
         self._kwargs_from_switches(_allowed_keys(runner_alias)),
         self.job_runner_kwargs(),
     )

 def _runner_kwargs(self):
     # just use combine_dicts() and not combine_confs(); leave the
     # magic to the runner
     return combine_dicts(
         self._non_option_kwargs(),
         # don't screen out irrelevant opts (see #1898)
         self._kwargs_from_switches(set(_RUNNER_OPTS)),
         self._job_kwargs(),
     )

    def test_can_override_sort_values_from_job(self):
        mr_job = MRSortValuesAndMore()

        self.assertEqual(
            mr_job.partitioner(),
            'org.apache.hadoop.mapred.lib.HashPartitioner')

        self.assertEqual(
            mr_job.jobconf(),
            combine_dicts(_SORT_VALUES_JOBCONF, MRSortValuesAndMore.JOBCONF))

    def _jobconf_for_step(self, step_num):
        """Get the jobconf dictionary, optionally including step-specific
        jobconf info.

        Also translate jobconfs to the current Hadoop version, if necessary.
        """
        step = self._get_step(step_num)
        jobconf = combine_dicts(self._opts["jobconf"], step.get("jobconf"))

        return add_translated_jobconf_for_hadoop_version(jobconf, self.get_hadoop_version())

    def hadoop_job_runner_kwargs(self):
        """Keyword arguments to create create runners when
        :py:meth:`make_runner` is called, when we run a job on EMR
        (``-r hadoop``).

        :return: map from arg name to value

        Re-define this if you want finer control when running jobs on hadoop.
        """
        return combine_dicts(self.job_runner_kwargs(), self._get_kwargs_from_opt_group(self.hadoop_opt_group))

    def _hadoop_conf_args(self, step, step_num, num_steps):
        """Build a list of extra arguments to the hadoop binary.

        This handles *cmdenv*, *hadoop_extra_args*, *hadoop_input_format*,
        *hadoop_output_format*, *jobconf*, and *partitioner*.

        This doesn't handle input, output, mappers, reducers, or uploading
        files.
        """
        assert 0 <= step_num < num_steps

        args = []

        jobconf = combine_dicts(self._opts['jobconf'], step.get('jobconf'))

        # hadoop_extra_args
        args.extend(self._opts['hadoop_extra_args'])

        # new-style jobconf
        version = self.get_hadoop_version()

        # translate the jobconf configuration names to match
        # the hadoop version
        jobconf = add_translated_jobconf_for_hadoop_version(jobconf,
                                                            version)
        if uses_generic_jobconf(version):
            for key, value in sorted(jobconf.iteritems()):
                if value is not None:
                    args.extend(['-D', '%s=%s' % (key, value)])
        # old-style jobconf
        else:
            for key, value in sorted(jobconf.iteritems()):
                if value is not None:
                    args.extend(['-jobconf', '%s=%s' % (key, value)])

        # partitioner
        if self._partitioner:
            args.extend(['-partitioner', self._partitioner])

        # cmdenv
        for key, value in sorted(self._opts['cmdenv'].iteritems()):
            args.append('-cmdenv')
            args.append('%s=%s' % (key, value))

        # hadoop_input_format
        if (step_num == 0 and self._hadoop_input_format):
            args.extend(['-inputformat', self._hadoop_input_format])

        # hadoop_output_format
        if (step_num == num_steps - 1 and self._hadoop_output_format):
            args.extend(['-outputformat', self._hadoop_output_format])

        return args

    def test_can_override_sort_values_from_cmd_line(self):
        mr_job = MRSortValues(
            ['--partitioner', 'org.pants.FancyPantsPartitioner',
             '--jobconf', 'stream.num.map.output.key.fields=lots'])

        self.assertEqual(
            mr_job.partitioner(),
            'org.pants.FancyPantsPartitioner')

        self.assertEqual(
            mr_job.jobconf(),
            combine_dicts(_SORT_VALUES_JOBCONF,
                          {'stream.num.map.output.key.fields': 'lots'}))

    def _jobconf_for_step(self, step_num):
        """Get the jobconf dictionary, optionally including step-specific
        jobconf info.

        Also translate jobconfs to the current Hadoop version, if necessary.
        """
        step = self._get_step(step_num)
        jobconf = combine_dicts(self._opts["jobconf"], step.get("jobconf"))

        # if user is using the wrong jobconfs, add in the correct ones
        self._update_jobconf_for_hadoop_version(jobconf, self.get_hadoop_version())

        return jobconf