Returns an array of each intermediate stage of a call to a `reduce`-like function.
970747
function(array, fun, init) {
      var ret = [];
      var acc = init;

      _.each(array, function(v,k) {
        acc = fun(acc, array[k]);
        ret.push(acc);
      });

      return ret;
    }
745690
def downstream(self, step_name):
        """"""
        return list(self.steps[dep] for dep in self.dag.downstream(step_name))
715695
def reduce(function, initval=None):
	
	if initval is None:
		return lambda s: __builtin__.reduce(function, s)
	else:
		return lambda s: __builtin__.reduce(function, s, initval)
451446
def to_array(self):
        
        stepdef = {'keep': self._keep,
                   'language': self._language,
                   'arg': self._arg}

        if self._language == 'javascript':
            if isinstance(self._function, list):
                stepdef['bucket'] = self._function[0]
                stepdef['key'] = self._function[1]
            elif isinstance(self._function, string_types):
                if ("{" in self._function):
                    stepdef['source'] = self._function
                else:
                    stepdef['name'] = self._function

        elif (self._language == 'erlang' and isinstance(self._function, list)):
            stepdef['module'] = self._function[0]
            stepdef['function'] = self._function[1]

        elif (self._language == 'erlang' and
              isinstance(self._function, string_types)):
            stepdef['source'] = self._function

        return {self._type: stepdef}
348759
private function sort(array $stageSteps)
    {
        if (empty($stageSteps)) {
            return [];
        }

        $dependencyGraphBuilder = new StepDependencyGraphBuilder($stageSteps);
        return $dependencyGraphBuilder->build()->sort();
    }
874189
private List<Instruction> reduceInstructionsInternal(final List<Instruction> instructions) {
        final List<Instruction> visitedInstructions = new LinkedList<>();
        final Set<Integer> visitedInstructionPositions = new HashSet<>();
        final Set<Integer> handledLoadIndexes = new HashSet<>();
        final Set<Integer> backtrackPositions = new LinkedHashSet<>(findSortedBacktrackPositions());

        while (!visitedInstructionPositions.containsAll(backtrackPositions)) {

            // unvisited backtrack position
            final int backtrackPosition = backtrackPositions.stream().filter(pos -> !visitedInstructionPositions.contains(pos))
                    .findFirst().orElseThrow(IllegalStateException::new);

            final List<Integer> lastVisitedPositions = stackSizeSimulator.simulateStatementBackwards(backtrackPosition);
            final List<Instruction> lastVisitedInstructions = lastVisitedPositions.stream().map(instructions::get).collect(Collectors.toList());

            visitedInstructionPositions.addAll(lastVisitedPositions);
            visitedInstructions.addAll(lastVisitedInstructions);

            // unhandled load indexes
            final Set<Integer> unhandledLoadIndexes = findUnhandledLoadIndexes(handledLoadIndexes, lastVisitedInstructions);

            // for each load occurrence index -> find load/store backtrack positions (reverse order matters here)
            final SortedSet<Integer> loadStoreBacktrackPositions = findLoadStoreBacktrackPositions(unhandledLoadIndexes);

            handledLoadIndexes.addAll(unhandledLoadIndexes);

            loadStoreBacktrackPositions.stream().forEach(backtrackPositions::add);
        }

        // sort in method natural order
        Collections.reverse(visitedInstructions);

        return visitedInstructions;
    }
588404
function reduceFunctionCall(string, functionRE, callback) {
  var call = string
  return getFunctionCalls(string, functionRE).reduce(function(string, obj) {
    return string.replace(obj.functionIdentifier + "(" + obj.matches.body + ")", evalFunctionCall(obj.matches.body, obj.functionIdentifier, callback, call, functionRE))
  }, string)
}
955404
def _reduced_call(self, **params_to_override):
        
        p=param.ParamOverrides(self,params_to_override)

        fn_result = self.function(p)
        self._apply_mask(p,fn_result)
        result = p.scale*fn_result+p.offset
        return result
239068
def _find_stages(self):
        """"""
        stages = []
        end = last_user_found = None
        for part in reversed(self.dfp.structure):
            if end is None:
                end = part

            if part['instruction'] == 'USER' and not last_user_found:
                # we will reuse the line verbatim
                last_user_found = part['content']

            if part['instruction'] == 'FROM':
                stages.insert(0, {'from_structure': part,
                                  'end_structure': end,
                                  'stage_user': last_user_found})
                end = last_user_found = None

        return stages
881063
def _summarize_call(parts):
    
    svtype = [x.split("=")[1] for x in parts[7].split(";") if x.startswith("SVTYPE=")]
    svtype = svtype[0] if svtype else ""
    start, end = _get_start_end(parts)
    return {"svtype": svtype, "size": int(end) - int(start)}
278394
protected function getSanitizedCalls()
    {
        $calls = $this->getCalls();

        foreach ($calls as &$call) {
            foreach ($call['arguments'] as &$argument) {
                if ($argument instanceof \Closure) {
                    $argument = Serialize::encodeClosure($argument);
                }
            }
        }

        return $calls;
    }
1330004
private void collectStartVertices(final ExecutionStage stage) {

		for (int i = 0; i < stage.getNumberOfStageMembers(); i++) {

			final ExecutionGroupVertex groupVertex = stage.getStageMember(i);

			if (forward) {
				if ((groupVertex.getNumberOfBackwardLinks() == 0)
					|| ((this.stage >= 0) && allConnectionsFromOtherStage(groupVertex, true))) {
					this.entryVertices.add(groupVertex);
				}
			} else {
				if ((groupVertex.getNumberOfForwardLinks() == 0)
					|| ((this.stage >= 0) && allConnectionsFromOtherStage(groupVertex, false))) {
					this.entryVertices.add(groupVertex);
				}
			}
		}
	}
38749
def getLeaves(self):
        """"""
        result = list()
        if not self._next_stages:
            result.append(self)
        else:
            for stage in self._next_stages:
                leaves = stage.getLeaves()
                result += leaves
        return result
843000
def get_stages(self):
        '''
        
        '''
        sequence = self.get_sequence()
        stages = {}
        for seq in self._sequences:
            for e in seq:
                if e.src.get_name() not in stages:
                    stages[e.src.get_name()] = []
                if e.dst.get_name() not in stages[e.src.get_name()]:
                    stages[e.src.get_name()].append(e.dst.get_name())
        return {
            'current': [e.dst.get_name() for e in sequence],
            'stages': stages
        }
1048885
def get_chunks(Array, Chunksize):
    """"""
    for i in range(0, len(Array), Chunksize):
        yield Array[i:i + Chunksize]
729812
def _apply_all_time_reductions(self, data):
        """"""
        logging.info(self._print_verbose("Applying desired time-"
                                         "reduction methods."))
        reduc_specs = [r.split('.') for r in self.dtype_out_time]
        reduced = {}
        for reduc, specs in zip(self.dtype_out_time, reduc_specs):
            func = specs[-1]
            if 'reg' in specs:
                reduced.update({reduc: self.region_calcs(data, func)})
            else:
                reduced.update({reduc: self._time_reduce(data, func)})
        return OrderedDict(sorted(reduced.items(), key=lambda t: t[0]))
1349991
def pipeline(steps, initial=None):
    

    def apply(result, step):
        return step(result)

    return reduce(apply, steps, initial)