#!/usr/bin/env python # # Copyright 2007 Google Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # """Parameters to control Mapreduce.""" __all__ = ["CONFIG_NAMESPACE", "config"] import pickle import google try: from appengine_pipeline.src.pipeline import util as pipeline_util except ImportError: pipeline_util = None from google.appengine.api import lib_config CONFIG_NAMESPACE = "mapreduce" DYNAMIC_RATE_INITIAL_QPS_PARAM = "initial_qps" DYNAMIC_RATE_BUMP_FACTOR_PARAM = "bump_factor" DYNAMIC_RATE_BUMP_TIME_PARAM = "bump_time" class _JobConfigMeta(type): """Metaclass that controls class creation.""" _OPTIONS = "_options" _REQUIRED = "_required" def __new__(mcs, classname, bases, class_dict): """Creates a _Config class and modifies its class dict. Args: classname: name of the class. bases: a list of base classes. class_dict: original class dict. Returns: A new _Config class. The modified class will have two fields. _options field is a dict from option name to _Option objects. _required field is a set of required option names. """ options = {} required = set() for name, option in class_dict.iteritems(): if isinstance(option, _Option): options[name] = option if option.required: required.add(name) for name in options: class_dict.pop(name) class_dict[mcs._OPTIONS] = options class_dict[mcs._REQUIRED] = required cls = type.__new__(mcs, classname, bases, class_dict) if object not in bases: parent_options = {} for c in reversed(cls.__mro__): if mcs._OPTIONS in c.__dict__: parent_options.update(c.__dict__[mcs._OPTIONS]) if mcs._REQUIRED in c.__dict__: required.update(c.__dict__[mcs._REQUIRED]) for k, v in parent_options.iteritems(): if k not in options: options[k] = v return cls class _Option(object): """An option for _Config.""" def __init__(self, kind, required=False, default_factory=None, can_be_none=False): """Init. Args: kind: type of the option. required: whether user is required to supply a value. default_factory: a factory, when called, returns the default value. can_be_none: whether value can be None. Raises: ValueError: if arguments aren't compatible. """ if required and default_factory is not None: raise ValueError("No default_factory value when option is required.") self.kind = kind self.required = required self.default_factory = default_factory self.can_be_none = can_be_none class _Config(object): """Root class for all per job configuration.""" __metaclass__ = _JobConfigMeta def __init__(self, _lenient=False, **kwds): """Init. Args: _lenient: When true, no option is required. **kwds: keyword arguments for options and their values. """ self._verify_keys(kwds, _lenient) self._set_values(kwds, _lenient) def _verify_keys(self, kwds, _lenient): keys = set() for k in kwds: if k not in self._options: raise ValueError("Option %s is not supported." % (k)) keys.add(k) if not _lenient: missing = self._required - keys if missing: raise ValueError("Options %s are required." % tuple(missing)) def _set_values(self, kwds, _lenient): for k, option in self._options.iteritems(): v = kwds.get(k) if v is None and option.default_factory: v = option.default_factory() setattr(self, k, v) if _lenient: continue if v is None and option.can_be_none: continue if isinstance(v, type) and not issubclass(v, option.kind): raise TypeError( "Expect subclass of %r for option %s. Got %r" % ( option.kind, k, v)) if not isinstance(v, type) and not isinstance(v, option.kind): raise TypeError("Expect type %r for option %s. Got %r" % ( option.kind, k, v)) def __eq__(self, other): if not isinstance(other, self.__class__): return False return other.__dict__ == self.__dict__ def __repr__(self): return str(self.__dict__) def to_json(self): return {"config": pickle.dumps(self)} @classmethod def from_json(cls, json): return pickle.loads(json["config"]) class _ConfigDefaults(object): """Default configs. Do not change parameters whose names begin with _. SHARD_MAX_ATTEMPTS: Max attempts to execute a shard before giving up. TASK_MAX_ATTEMPTS: Max attempts to execute a task before dropping it. Task is any taskqueue task created by MR framework. A task is dropped when its X-AppEngine-TaskExecutionCount is bigger than this number. Dropping a task will cause abort on the entire MR job. TASK_MAX_DATA_PROCESSING_ATTEMPTS: Max times to execute a task when previous task attempts failed during data processing stage. An MR work task has three major stages: initial setup, data processing, and final checkpoint. Setup stage should be allowed to be retried more times than data processing stage: setup failures are caused by unavailable GAE services while data processing failures are mostly due to user function error out on certain input data. Thus, set TASK_MAX_ATTEMPTS higher than this parameter. QUEUE_NAME: Default queue for MR. SHARD_COUNT: Default shard count. INITIAL_QPS, BUMP_FACTOR, BUMP_TIME: Allows slow MR ramp up (to avoid hotspotting datastore). We start at initial_qps and increase the pace by bump_factor each bump_time seconds. So for initial_qps=500, bump_factor=1.5 and bump_time=300 we would start with 500qps (per MR) and increase it by 50% every 5min. We do continuous increases. If set (all 3 non-zero) takes precedence over PROCESSING_RATE_PER_SEC. PROCESSING_RATE_PER_SEC: Default rate of processed entities per second. BASE_PATH : Base path of mapreduce and pipeline handlers. """ SHARD_MAX_ATTEMPTS = 4 TASK_MAX_ATTEMPTS = 31 TASK_MAX_DATA_PROCESSING_ATTEMPTS = 11 QUEUE_NAME = "default" SHARD_COUNT = 8 INITIAL_QPS = 0 BUMP_FACTOR = 0 BUMP_TIME = 0 PROCESSING_RATE_PER_SEC = 1000000 BASE_PATH = "/_ah/mapreduce" PERSIST_GCS_ACCESS_TOKEN = True _SLICE_DURATION_SEC = 15 _CONTROLLER_PERIOD_SEC = 2 config = lib_config.register(CONFIG_NAMESPACE, _ConfigDefaults.__dict__) _DEFAULT_PIPELINE_BASE_PATH = config.BASE_PATH + "/pipeline" _GCS_URLFETCH_TIMEOUT_SEC = 30 _LEASE_DURATION_SEC = config._SLICE_DURATION_SEC * 1.1 _MAX_LEASE_DURATION_SEC = max(10 * 60 + 30, config._SLICE_DURATION_SEC * 1.5)