5d16d5bbfd
This pull request aims to resolve all outstanding PEP8 violations in PySpark. Author: Nicholas Chammas <nicholas.chammas@gmail.com> Author: nchammas <nicholas.chammas@gmail.com> Closes #1505 from nchammas/master and squashes the following commits: 98171af [Nicholas Chammas] [SPARK-2470] revert PEP 8 fixes to cloudpickle cba7768 [Nicholas Chammas] [SPARK-2470] wrap expression list in parentheses e178dbe [Nicholas Chammas] [SPARK-2470] style - change position of line break 9127d2b [Nicholas Chammas] [SPARK-2470] wrap expression lists in parentheses 22132a4 [Nicholas Chammas] [SPARK-2470] wrap conditionals in parentheses 24639bc [Nicholas Chammas] [SPARK-2470] fix whitespace for doctest 7d557b7 [Nicholas Chammas] [SPARK-2470] PEP8 fixes to tests.py 8f8e4c0 [Nicholas Chammas] [SPARK-2470] PEP8 fixes to storagelevel.py b3b96cf [Nicholas Chammas] [SPARK-2470] PEP8 fixes to statcounter.py d644477 [Nicholas Chammas] [SPARK-2470] PEP8 fixes to worker.py aa3a7b6 [Nicholas Chammas] [SPARK-2470] PEP8 fixes to sql.py 1916859 [Nicholas Chammas] [SPARK-2470] PEP8 fixes to shell.py 95d1d95 [Nicholas Chammas] [SPARK-2470] PEP8 fixes to serializers.py a0fec2e [Nicholas Chammas] [SPARK-2470] PEP8 fixes to mllib c85e1e5 [Nicholas Chammas] [SPARK-2470] PEP8 fixes to join.py d14f2f1 [Nicholas Chammas] [SPARK-2470] PEP8 fixes to __init__.py 81fcb20 [Nicholas Chammas] [SPARK-2470] PEP8 fixes to resultiterable.py 1bde265 [Nicholas Chammas] [SPARK-2470] PEP8 fixes to java_gateway.py 7fc849c [Nicholas Chammas] [SPARK-2470] PEP8 fixes to daemon.py ca2d28b [Nicholas Chammas] [SPARK-2470] PEP8 fixes to context.py f4e0039 [Nicholas Chammas] [SPARK-2470] PEP8 fixes to conf.py a6d5e4b [Nicholas Chammas] [SPARK-2470] PEP8 fixes to cloudpickle.py f0a7ebf [Nicholas Chammas] [SPARK-2470] PEP8 fixes to rddsampler.py 4dd148f [nchammas] Merge pull request #5 from apache/master f7e4581 [Nicholas Chammas] unrelated pep8 fix a36eed0 [Nicholas Chammas] name ec2 instances and security groups consistently de7292a [nchammas] Merge pull request #4 from apache/master 2e4fe00 [nchammas] Merge pull request #3 from apache/master 89fde08 [nchammas] Merge pull request #2 from apache/master 69f6e22 [Nicholas Chammas] PEP8 fixes 2627247 [Nicholas Chammas] broke up lines before they hit 100 chars 6544b7e [Nicholas Chammas] [SPARK-2065] give launched instances names 69da6cf [nchammas] Merge pull request #1 from apache/master
110 lines
3.8 KiB
Python
110 lines
3.8 KiB
Python
#
|
|
# Licensed to the Apache Software Foundation (ASF) under one or more
|
|
# contributor license agreements. See the NOTICE file distributed with
|
|
# this work for additional information regarding copyright ownership.
|
|
# The ASF licenses this file to You 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.
|
|
#
|
|
|
|
import sys
|
|
import random
|
|
|
|
|
|
class RDDSampler(object):
|
|
def __init__(self, withReplacement, fraction, seed=None):
|
|
try:
|
|
import numpy
|
|
self._use_numpy = True
|
|
except ImportError:
|
|
print >> sys.stderr, (
|
|
"NumPy does not appear to be installed. "
|
|
"Falling back to default random generator for sampling.")
|
|
self._use_numpy = False
|
|
|
|
self._seed = seed if seed is not None else random.randint(0, sys.maxint)
|
|
self._withReplacement = withReplacement
|
|
self._fraction = fraction
|
|
self._random = None
|
|
self._split = None
|
|
self._rand_initialized = False
|
|
|
|
def initRandomGenerator(self, split):
|
|
if self._use_numpy:
|
|
import numpy
|
|
self._random = numpy.random.RandomState(self._seed)
|
|
else:
|
|
self._random = random.Random(self._seed)
|
|
|
|
for _ in range(0, split):
|
|
# discard the next few values in the sequence to have a
|
|
# different seed for the different splits
|
|
self._random.randint(0, sys.maxint)
|
|
|
|
self._split = split
|
|
self._rand_initialized = True
|
|
|
|
def getUniformSample(self, split):
|
|
if not self._rand_initialized or split != self._split:
|
|
self.initRandomGenerator(split)
|
|
|
|
if self._use_numpy:
|
|
return self._random.random_sample()
|
|
else:
|
|
return self._random.uniform(0.0, 1.0)
|
|
|
|
def getPoissonSample(self, split, mean):
|
|
if not self._rand_initialized or split != self._split:
|
|
self.initRandomGenerator(split)
|
|
|
|
if self._use_numpy:
|
|
return self._random.poisson(mean)
|
|
else:
|
|
# here we simulate drawing numbers n_i ~ Poisson(lambda = 1/mean) by
|
|
# drawing a sequence of numbers delta_j ~ Exp(mean)
|
|
num_arrivals = 1
|
|
cur_time = 0.0
|
|
|
|
cur_time += self._random.expovariate(mean)
|
|
|
|
if cur_time > 1.0:
|
|
return 0
|
|
|
|
while(cur_time <= 1.0):
|
|
cur_time += self._random.expovariate(mean)
|
|
num_arrivals += 1
|
|
|
|
return (num_arrivals - 1)
|
|
|
|
def shuffle(self, vals):
|
|
if self._random is None:
|
|
self.initRandomGenerator(0) # this should only ever called on the master so
|
|
# the split does not matter
|
|
|
|
if self._use_numpy:
|
|
self._random.shuffle(vals)
|
|
else:
|
|
self._random.shuffle(vals, self._random.random)
|
|
|
|
def func(self, split, iterator):
|
|
if self._withReplacement:
|
|
for obj in iterator:
|
|
# For large datasets, the expected number of occurrences of each element in
|
|
# a sample with replacement is Poisson(frac). We use that to get a count for
|
|
# each element.
|
|
count = self.getPoissonSample(split, mean=self._fraction)
|
|
for _ in range(0, count):
|
|
yield obj
|
|
else:
|
|
for obj in iterator:
|
|
if self.getUniformSample(split) <= self._fraction:
|
|
yield obj
|