<commit_before>import os, os.path
import time
import threading
import socket
import csv
import cStringIO
import itertools
import operator

from serialize import load_func, dump_func
from dependency import *
from env import env
import moosefs

def ilen(x):
    try:
        return len(x)
    except TypeError:
        return sum(1 for i in x)

class Split:
    def __init__(self, idx):
        self.index = idx

def cached(func):
    def getstate(self):
        d = self._pickle_cache.get(self.id)
        if d is None:
            d = func(self)
            self._pickle_cache[self.id] = d
        return d
    return getstate

class RDD:
    _pickle_cache = {}
    def __init__(self, ctx):
        self.ctx = ctx
        self.id = RDD.newId()
        self._splits = []
        self.dependencies = []
        self.aggregator = None
        self._partitioner = None
        self.shouldCache = False

    nextId = 0
    @classmethod
    def newId(cls):
        cls.nextId += 1
        return cls.nextId

    @cached
    def __getstate__(self):
        d = dict(self.__dict__)
        d.pop('dependencies', None)
        d.pop('_splits', None)
        d.pop('ctx', None)
        return d

    def __len__(self):
        return len(self.splits)
   
    def __getslice__(self, i,j):
        return SliceRDD(self, i, j)

    @property
    def splits(self):
        return self._splits

    def compute(self, split):
        raise NotImplementedError

    @property
    def partitioner(self):
        return self._partitioner
    
    def preferredLocations(self, split):
        return []

    def cache(self):
        self.shouldCache = True
        return self

    def iterator(self, split):
        if self.shouldCache:
            return env.cacheTracker.getOrCompute(self, split)
        else:
            return self.compute(split)

    def map(self, f):
        return MappedRDD(self, f)

    def flatMap(self, f):
        return FlatMappedRDD(self, f)

    def filter(self, f):
        return FilteredRDD(self, f)

    def sample(self, withReplacement, faction, seed):
        return SampleRDD(self, withReplacement, faction, seed)

    def union(self, rdd):
        return UnionRDD(self.ctx, [self, rdd])

    def sort(self, key=lambda x:x, numSplits=None):
        if numSplits is None:
            numSplits = min(self.ctx.defaultMinSplits, len(self))
        n = numSplits * 10 / len(self)
        samples = self.glom().flatMap(lambda x:itertools.islice(x, n)).map(key).collect()
        keys = sorted(samples)[5::10][:numSplits-1]
        aggr = MergeAggregator()
        parter = RangePartitioner(keys)
        parted = ShuffledRDD(self.map(lambda x:(key(x),x)), aggr, parter).flatMap(lambda (x,y):y)
        return parted.glom().flatMap(lambda x:sorted(x))

    def glom(self):
        return GlommedRDD(self)

    def cartesion(self, other):
        return CartesionRDD(self, other)

    def groupBy(self, f, numSplits=None):
        if numSplits is None:
            numSplits = min(self.ctx.defaultMinSplits, len(self))
        return self.map(lambda x: (f(x), x)).groupByKey(numSplits)

    def pipe(self, command):
        if isinstance(command, str):
            command = command.split(' ')
        return PipedRDD(self, command)

    def mapPartitions(self, f):
        return MapPartitionsRDD(self, f)

    def foreach(self, f):
        def mf(it):
            for i in it:
                f(i)
        return self.ctx.runJob(self, mf)

    def collect(self):
        return sum(self.ctx.runJob(self, lambda x:list(x)), [])

    def __iter__(self):
        return self.collect()

    def reduce(self, f):
        def reducePartition(it):
            if it:
                return [reduce(f, it)]
            else:
                return []
        options = self.ctx.runJob(self, reducePartition)
        return reduce(f, sum(options, []))
<commit_msg>bugfix: reduce with empty seq<commit_after>import os, os.path
import time
import threading
import socket
import csv
import cStringIO
import itertools
import operator

from serialize import load_func, dump_func
from dependency import *
from env import env
import moosefs

def ilen(x):
    try:
        return len(x)
    except TypeError:
        return sum(1 for i in x)

class Split:
    def __init__(self, idx):
        self.index = idx

def cached(func):
    def getstate(self):
        d = self._pickle_cache.get(self.id)
        if d is None:
            d = func(self)
            self._pickle_cache[self.id] = d
        return d
    return getstate

class RDD:
    _pickle_cache = {}
    def __init__(self, ctx):
        self.ctx = ctx
        self.id = RDD.newId()
        self._splits = []
        self.dependencies = []
        self.aggregator = None
        self._partitioner = None
        self.shouldCache = False

    nextId = 0
    @classmethod
    def newId(cls):
        cls.nextId += 1
        return cls.nextId

    @cached
    def __getstate__(self):
        d = dict(self.__dict__)
        d.pop('dependencies', None)
        d.pop('_splits', None)
        d.pop('ctx', None)
        return d

    def __len__(self):
        return len(self.splits)
   
    def __getslice__(self, i,j):
        return SliceRDD(self, i, j)

    @property
    def splits(self):
        return self._splits

    def compute(self, split):
        raise NotImplementedError

    @property
    def partitioner(self):
        return self._partitioner
    
    def preferredLocations(self, split):
        return []

    def cache(self):
        self.shouldCache = True
        return self

    def iterator(self, split):
        if self.shouldCache:
            return env.cacheTracker.getOrCompute(self, split)
        else:
            return self.compute(split)

    def map(self, f):
        return MappedRDD(self, f)

    def flatMap(self, f):
        return FlatMappedRDD(self, f)

    def filter(self, f):
        return FilteredRDD(self, f)

    def sample(self, withReplacement, faction, seed):
        return SampleRDD(self, withReplacement, faction, seed)

    def union(self, rdd):
        return UnionRDD(self.ctx, [self, rdd])

    def sort(self, key=lambda x:x, numSplits=None):
        if numSplits is None:
            numSplits = min(self.ctx.defaultMinSplits, len(self))
        n = numSplits * 10 / len(self)
        samples = self.glom().flatMap(lambda x:itertools.islice(x, n)).map(key).collect()
        keys = sorted(samples)[5::10][:numSplits-1]
        aggr = MergeAggregator()
        parter = RangePartitioner(keys)
        parted = ShuffledRDD(self.map(lambda x:(key(x),x)), aggr, parter).flatMap(lambda (x,y):y)
        return parted.glom().flatMap(lambda x:sorted(x))

    def glom(self):
        return GlommedRDD(self)

    def cartesion(self, other):
        return CartesionRDD(self, other)

    def groupBy(self, f, numSplits=None):
        if numSplits is None:
            numSplits = min(self.ctx.defaultMinSplits, len(self))
        return self.map(lambda x: (f(x), x)).groupByKey(numSplits)

    def pipe(self, command):
        if isinstance(command, str):
            command = command.split(' ')
        return PipedRDD(self, command)

    def mapPartitions(self, f):
        return MapPartitionsRDD(self, f)

    def foreach(self, f):
        def mf(it):
            for i in it:
                f(i)
        return self.ctx.runJob(self, mf)

    def collect(self):
        return sum(self.ctx.runJob(self, lambda x:list(x)), [])

    def __iter__(self):
        return self.collect()

    def reduce(self, f):
        def reducePartition(it):
            try:
                return [reduce(f, it)]
            except TypeError:
                return []
        options = self.ctx.runJob(self, reducePartition)
        s = sum(options, [])
        if s:
            return reduce(f, s)

{'fraction_non_alphanumeric': 0.10884615384615384, 'fraction_numerical': 0.002307692307692308, 'mean_word_length': 2.181484502446982, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1397573', 'n_tokens_mistral': 2545, 'n_tokens_neox': 2444, 'n_words': 628}