The base class for all MapReduce jobs. See __init__() for details.
Re-define this to define the mapper for a one-step job.
Yields zero or more tuples of (out_key, out_value).
| Parameters: |
|
|---|
If you don’t re-define this, your job will have a mapper that simply yields (key, value) as-is.
Re-define this to define the reducer for a one-step job.
Yields one or more tuples of (out_key, out_value)
| Parameters: |
|
|---|
Re-define this to define the combiner for a one-step job.
Yields one or more tuples of (out_key, out_value)
| Parameters: |
|
|---|
Re-define this to define an action to run before the mapper processes any input.
One use for this function is to initialize mapper-specific helper structures.
Yields one or more tuples of (out_key, out_value).
By default, out_key and out_value must be JSON-encodable; re-define INTERNAL_PROTOCOL to change this.
Re-define this to define an action to run after the mapper reaches the end of input.
One way to use this is to store a total in an instance variable, and output it after reading all input data. See mrjob.examples for an example.
Yields one or more tuples of (out_key, out_value).
By default, out_key and out_value must be JSON-encodable; re-define INTERNAL_PROTOCOL to change this.
Re-define this to define an action to run before the reducer processes any input.
One use for this function is to initialize reducer-specific helper structures.
Yields one or more tuples of (out_key, out_value).
By default, out_key and out_value must be JSON-encodable; re-define INTERNAL_PROTOCOL to change this.
Re-define this to define an action to run after the reducer reaches the end of input.
Yields one or more tuples of (out_key, out_value).
By default, out_key and out_value must be JSON-encodable; re-define INTERNAL_PROTOCOL to change this.
Re-define this to define an action to run before the combiner processes any input.
One use for this function is to initialize combiner-specific helper structures.
Yields one or more tuples of (out_key, out_value).
By default, out_key and out_value must be JSON-encodable; re-define INTERNAL_PROTOCOL to change this.
Re-define this to define an action to run after the combiner reaches the end of input.
Yields one or more tuples of (out_key, out_value).
By default, out_key and out_value must be JSON-encodable; re-define INTERNAL_PROTOCOL to change this.
Re-define this to make a multi-step job.
If you don’t re-define this, we’ll automatically create a one-step job using any of mapper(), mapper_init(), mapper_final(), reducer_init(), reducer_final(), and reducer() that you’ve re-defined. For example:
def steps(self):
return [self.mr(mapper=self.transform_input,
reducer=self.consolidate_1),
self.mr(reducer_init=self.log_mapper_init,
reducer=self.consolidate_2)]
| Returns: | a list of steps constructed with mr() |
|---|
Define a step (mapper, reducer, and/or any combination of mapper_init, reducer_final, etc.) for your job.
Used by steps(). (Don’t re-define this, just call it!)
Accepts the following keyword arguments. For convenience, you may specify mapper and reducer as positional arguments as well.
| Parameters: |
|
|---|
Please consider the way we represent steps to be opaque, and expect it to change in future versions of mrjob.
Entry point for running job from the command-line.
This is also the entry point when a mapper or reducer is run by Hadoop Streaming.
Does one of:
Entry point for running your job from other Python code.
You can pass in command-line arguments, and the job will act the same way it would if it were run from the command line. For example, to run your job on EMR:
mr_job = MRYourJob(args=['-r', 'emr'])
with mr_job.make_runner() as runner:
...
Passing in None is the same as passing in [] (if you want to parse args from sys.argv, call MRJob.run()).
For a full list of command-line arguments, run: python -m mrjob.job --help
Make a runner based on command-line arguments, so we can launch this job on EMR, on Hadoop, or locally.
| Return type: | mrjob.runner.MRJobRunner |
|---|
Parse a line from the final output of this MRJob into (key, value). Used extensively in tests like this:
runner.run()
for line in runner.stream_output():
key, value = mr_job.parse_output_line(line)
Increment a counter in Hadoop streaming by printing to stderr.
| Parameters: |
|
|---|
Commas in counter or group will be automatically replaced with semicolons (commas confuse Hadoop streaming).
Set the job status in hadoop streaming by printing to stderr.
This is also a good way of doing a keepalive for a job that goes a long time between outputs; Hadoop streaming usually times out jobs that give no output for longer than 10 minutes.
Protocols convert data between string representations for Hadoop Streaming and Python data structures for input and output by steps. Using the default protocols, the first step will get raw strings as input and should return JSON-encodable Python objects. The last step should accept JSON-decoded Python objects and return JSON-encodable Python objects, which will appear in the final output as JSON.
There are three ways to specify which protocol you want to use for a given step. The simplest way is to assign a class imported from mrjob.protocols to one of the class variables MRJob.INPUT_PROTOCOL, MRJob.INTERNAL_PROTOCOL, or MRJob.OUTPUT_PROTOCOL on your job. The input protocol is the format of your input data, the internal protocol is the format of data passed between steps, and the output protocol is what is produced at the end of the job. These classes are instantiated when your MRJob is instantiated.
For example, this class accepts raw strings, outputs JSON, and uses pickle to communicate between steps:
from mrjob.job import MRJob
from mrjob.protocol import JSONValueProtocol, PickleProtocol, RawValueProtocol
class MRProtocolJob1(MRJob)
INPUT_PROTOCOL = RawValueProtocol
INTERNAL_PROTOCOL = PickleProtocol
OUTPUT_PROTOCOL = JSONValueProtocol
The second way is to override MRJob.input_protocol(), MRJob.internal_protocol(), or MRJob.output_protocol(). These methods should return instances of protocols. For example, if you want your job to operate on either JSON or raw strings depending on a command line option, you can do this:
from mrjob.job import MRJob
from mrjob.protocol import JSONValueProtocol, RawValueProtocol
class MRProtocolJob2(MRJob):
def configure_options(self):
# set up 'data_format' option
def input_protocol(self):
if self.options.data_format == 'json':
return JSONValueProtocol()
elif self.options.data_format == 'raw':
return RawValueProtocol()
else:
raise ValueError("Data format must be 'json' or 'raw'")
If you need behavior even more complex than that, including using different protocols to communicate between different steps (like JSON between steps 1 and 2, but pickle between steps 2 and 3), you can override MRJob.pick_protocols() to return different protocols based on the step number and step type of the current process.
See mrjob.protocol - input and output for more information about protocols, including how to implement your own.
Protocol for reading input to the first mapper in your job. Default: RawValueProtocol.
For example you know your input data were in JSON format, you could set:
INPUT_PROTOCOL = JsonValueProtocol
in your class, and your initial mapper would receive decoded JSONs rather than strings.
See mrjob.protocol for the full list of protocols.
Protocol for communication between steps and final output. Default: JSONProtocol.
For example if your step output weren’t JSON-encodable, you could set:
INTERNAL_PROTOCOL = PickleProtocol
and step output would be encoded as string-escaped pickles.
See mrjob.protocol for the full list of protocols.
Protocol to use for writing output. Default: JSONProtocol.
For example, if you wanted the final output in repr, you could set:
OUTPUT_PROTOCOL = ReprProtocol
See mrjob.protocol for the full list of protocols.
Deprecated since version 0.3.0.
Default protocol for reading input to the first mapper in your job specified by a string.
Overridden by any changes to INPUT_PROTOCOL.
See mrjob.protocol.PROTOCOL_DICT for the full list of protocol strings. Can be overridden by --input-protocol.
Deprecated since version 0.3.0.
Default protocol for communication between steps and final output specified by a string.
Overridden by any changes to INTERNAL_PROTOCOL.
See mrjob.protocol.PROTOCOL_DICT for the full list of protocol strings. Can be overridden by --protocol.
Deprecated since version 0.3.0.
Overridden by any changes to OUTPUT_PROTOCOL. If OUTPUT_PROTOCOL is not set, defaults to DEFAULT_PROTOCOL.
See mrjob.protocol.PROTOCOL_DICT for the full list of protocol strings. Can be overridden by the --output-protocol.
Instance of the protocol to use to convert input lines to Python objects. Default behavior is to return an instance of INPUT_PROTOCOL.
Instance of the protocol to use to communicate between steps. Default behavior is to return an instance of INTERNAL_PROTOCOL.
Instance of the protocol to use to convert Python objects to output lines. Default behavior is to return an instance of OUTPUT_PROTOCOL.
Deprecated in favor of INPUT_PROTOCOL, OUTPUT_PROTOCOL, and INTERNAL_PROTOCOL.
Mapping from protocol name to the protocol class to use for parsing job input and writing job output. We give protocols names so that we can easily choose them from the command line.
This returns mrjob.protocol.PROTOCOL_DICT by default.
To add a custom protocol, define a subclass of mrjob.protocol.HadoopStreamingProtocol, and re-define this method:
@classmethod
def protocols(cls):
protocol_dict = super(MRYourJob, cls).protocols()
protocol_dict['rot13'] = Rot13Protocol
return protocol_dict
DEFAULT_PROTOCOL = 'rot13'
Pick the protocol classes to use for reading and writing for the given step.
| Parameters: |
|
|---|---|
| Returns: | (read_function, write_function) |
By default, we use one protocol for reading input, one internal protocol for communication between steps, and one protocol for final output (which is usually the same as the internal protocol). Protocols can be controlled by setting INPUT_PROTOCOL, INTERNAL_PROTOCOL, and OUTPUT_PROTOCOL.
Re-define this if you need fine control over which protocols are used by which steps.
See Quick Reference for a complete list of all configuration options.
Define arguments for this script. Called from __init__().
Run python -m mrjob.job.MRJob --help to see all options.
Re-define to define custom command-line arguments:
def configure_options(self):
super(MRYourJob, self).configure_options
self.add_passthrough_option(...)
self.add_file_option(...)
...
Function to create options which both the job runner and the job itself respect (we use this for protocols, for example).
Use it like you would use optparse.OptionParser.add_option():
def configure_options(self):
super(MRYourJob, self).configure_options()
self.add_passthrough_option(
'--max-ngram-size', type='int', default=4, help='...')
Specify an opt_group keyword argument to add the option to that OptionGroup rather than the top-level OptionParser.
If you want to pass files through to the mapper/reducer, use add_file_option() instead.
Add a command-line option that sends an external file (e.g. a SQLite DB) to Hadoop:
def configure_options(self):
super(MRYourJob, self).configure_options()
self.add_file_option('--scoring-db', help=...)
This does the right thing: the file will be uploaded to the working dir of the script on Hadoop, and the script will be passed the same option, but with the local name of the file in the script’s working directory.
We suggest against sending Berkeley DBs to your job, as Berkeley DB is not forwards-compatible (so a Berkeley DB that you construct on your computer may not be readable from within Hadoop). Use SQLite databases instead. If all you need is an on-disk hash table, try out the sqlite3dbm module.
Load command-line options into self.options.
Called from __init__() after configure_options().
| Parameters: | args (list of str) – a list of command line arguments. None will be treated the same as []. |
|---|
Re-define if you want to post-process command-line arguments:
def load_options(self, args):
super(MRYourJob, self).load_options(args)
self.stop_words = self.options.stop_words.split(',')
...
True if this is a mapper/reducer.
This is mostly useful inside load_options(), to disable loading options when we aren’t running inside Hadoop Streaming.
optparse.Option subclass to use with the optparse.OptionParser instance.
The optparse module allows the addition of new actions and types. See the optparse docs for instructions on defining custom options. The only difference is that instead of passing option_class to the OptionParser instance yourself, you must set the MRJob.OPTION_CLASS attribute.
Passthrough arguments have the additional caveat that mrjob uses some lesser magic to reproduce the argument values for the command lines of subprocesses. In practice you shouldn’t encounter any problems here even with relatively exotic option behavior, but be aware that your options will be processed twice, with the second round using a copy of your default values produced by copy.deepcopy().
Keyword arguments used to create runners when make_runner() is called.
| Returns: | map from arg name to value |
|---|
Re-define this if you want finer control of runner initialization.
You might find mrjob.conf.combine_dicts() useful if you want to add or change lots of keyword arguments.
Keyword arguments to create create runners when make_runner() is called, when we run a job locally (-r local).
| Returns: | map from arg name to value |
|---|
Re-define this if you want finer control when running jobs locally.
Keyword arguments to create create runners when make_runner() is called, when we run a job on EMR (-r emr).
| Returns: | map from arg name to value |
|---|
Re-define this if you want finer control when running jobs on EMR.
Keyword arguments to create create runners when make_runner() is called, when we run a job on EMR (-r hadoop).
| Returns: | map from arg name to value |
|---|
Re-define this if you want finer control when running jobs on hadoop.
Returns a list of arguments to pass to subprocesses, either on hadoop or executed via subprocess.
These are passed to mrjob.runner.MRJobRunner.__init__() as extra_args.
Figure out file upload args to pass through to the job runner.
Instead of generating a list of args, we’re generating a list of tuples of ('--argname', path)
These are passed to mrjob.runner.MRJobRunner.__init__() as file_upload_args.
Path of this script. This returns the file containing this class.
Run the all steps of the job, logging errors (and debugging output if --verbose is specified) to STDERR and streaming the output to STDOUT.
Called from run(). You’d probably only want to call this directly from automated tests.
Run the mapper and final mapper action for the given step.
| Parameters: | step_num (int) – which step to run (0-indexed) |
|---|
If we encounter a line that can’t be decoded by our input protocol, or a tuple that can’t be encoded by our output protocol, we’ll increment a counter rather than raising an exception. If –strict-protocols is set, then an exception is raised
Called from run(). You’d probably only want to call this directly from automated tests.
Run the reducer for the given step.
| Parameters: | step_num (int) – which step to run (0-indexed) |
|---|
If we encounter a line that can’t be decoded by our input protocol, or a tuple that can’t be encoded by our output protocol, we’ll increment a counter rather than raising an exception. If –strict-protocols is set, then an exception is raised
Called from run(). You’d probably only want to call this directly from automated tests.
Run the combiner for the given step.
| Parameters: | step_num (int) – which step to run (0-indexed) |
|---|
If we encounter a line that can’t be decoded by our input protocol, or a tuple that can’t be encoded by our output protocol, we’ll increment a counter rather than raising an exception. If –strict-protocols is set, then an exception is raised
Called from run(). You’d probably only want to call this directly from automated tests.
Print information about how many steps there are, and whether they contain a mapper or reducer. Job runners (see Runners - launching your job) use this to determine how Hadoop should call this script.
Called from run(). You’d probably only want to call this directly from automated tests.
We currently output something like MR M R, but expect this to change!
Optional name of an optional Hadoop InputFormat class, e.g. 'org.apache.hadoop.mapred.lib.NLineInputFormat'.
Passed to Hadoop with the first step of this job with the -inputformat option.
Optional Hadoop InputFormat class to parse input for the first step of the job.
Normally, setting HADOOP_INPUT_FORMAT is sufficient; redefining this method is only for when you want to get fancy.
Optional name of an optional Hadoop OutputFormat class, e.g. 'org.apache.hadoop.mapred.FileOutputFormat'.
Passed to Hadoop with the last step of this job with the -outputformat option.
Optional Hadoop OutputFormat class to write output for the last step of the job.
Normally, setting HADOOP_OUTPUT_FORMAT is sufficient; redefining this method is only for when you want to get fancy.
Optional jobconf arguments we should always pass to Hadoop. This is a map from property name to value. e.g.:
{'stream.num.map.output.key.fields': '4'}
It’s recommended that you only use this to hard-code things that affect the semantics of your job, and leave performance tweaks to the command line or whatever you use to launch your job.
-jobconf args to pass to hadoop streaming. This should be a map from property name to value.
By default, this combines jobconf options from the command lines with JOBCONF, with command line arguments taking precedence.
If you want to re-define this, it’s strongly recommended that do something like this, so as not to inadvertently disable jobconf:
def jobconf(self):
orig_jobconf = super(MyMRJobClass, self).jobconf()
custom_jobconf = ...
return mrjob.conf.combine_dicts(orig_jobconf, custom_jobconf)
Optional Hadoop partitioner class to use to determine how mapper output should be sorted and distributed to reducers. For example: 'org.apache.hadoop.mapred.lib.HashPartitioner'.
Optional Hadoop partitioner class to use to determine how mapper output should be sorted and distributed to reducers.
By default, returns whatever is passed to --partitioner, of if that option isn’t used, PARTITIONER.
You probably don’t need to re-define this; it’s just here for completeness.
Redirect stdin, stdout, and stderr for automated testing.
You can set stdin, stdout, and stderr to file objects. By default, they’ll be set to empty StringIO objects. You can then access the job’s file handles through self.stdin, self.stdout, and self.stderr. See Testing with mrjob for more information about testing.
You may call sandbox multiple times (this will essentially clear the file handles).
stdin is empty by default. You can set it to anything that yields lines:
mr_job.sandbox(stdin=StringIO('some_data\n'))
or, equivalently:
mr_job.sandbox(stdin=['some_data\n'])
For convenience, this sandbox() returns self, so you can do:
mr_job = MRJobClassToTest().sandbox()
Simple testing example:
mr_job = MRYourJob.sandbox()
assert_equal(list(mr_job.reducer('foo', ['bar', 'baz'])), [...])
More complex testing example:
from StringIO import StringIO
mr_job = MRYourJob(args=[...])
fake_input = '"foo"\t"bar"\n"foo"\t"baz"\n'
mr_job.sandbox(stdin=StringIO(fake_input))
mr_job.run_reducer(link_num=0)
assert_equal(mr_job.parse_output(), ...)
assert_equal(mr_job.parse_counters(), ...)
Convenience method for parsing output from any mapper or reducer, all at once.
This helps you test individual mappers and reducers by calling run_mapper() or run_reducer(). For example:
mr_job.sandbox(stdin=your_input)
mr_job.run_mapper(step_num=0)
output = mrjob.parse_output()
| Parameters: | protocol (str) – A protocol instance to use (e.g. JSONProtocol()), Also accepts protocol names (e.g. 'json'), but this is deprecated. |
|---|
This only works in sandbox mode. This does not clear self.stdout.
Convenience method for reading counters. This only works in sandbox mode. This does not clear self.stderr.
| Returns: | a map from counter group to counter name to amount. |
|---|
To read everything from self.stderr (including status messages) use mrjob.parse.parse_mr_job_stderr().
When writing unit tests, you may find MRJobRunner.counters() more useful.