mrjob with parameters

gistfile1.py

from mrjob.job import MRJob
from mrjob.protocol import JSONValueProtocol

class MRConversions(MRJob):

  def __init__(self, duration=1, offset=0):
    self.duration = duration
    self.offset = offset
    self.now = datetime.date.today()

  def mapper(self, key, record):
    if record.get('action') == 'send':
      content = record.get('content', None)
      created = record.get('created', None)
      if created >= self.now - datetime.timedelta(self.duration + self.offset) and created < self.now - datetime.timedelta(self.offset):
        yield '{}_{}'.format(content, record.get('user_type')), 1

  def reducer(self, key, pieces):
    yield key, sum(pieces)

if __name__ == '__main__':
  MRConversions.run(2, 5)