-
Notifications
You must be signed in to change notification settings - Fork 28.3k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
SPARK-1242 Add aggregate to python rdd #139
Changes from 1 commit
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -524,7 +524,7 @@ def _collect_iterator_through_file(self, iterator): | |
def reduce(self, f): | ||
""" | ||
Reduces the elements of this RDD using the specified commutative and | ||
associative binary operator. | ||
associative binary operator. Currently reduces partitions locally. | ||
|
||
>>> from operator import add | ||
>>> sc.parallelize([1, 2, 3, 4, 5]).reduce(add) | ||
|
@@ -566,8 +566,32 @@ def func(iterator): | |
vals = self.mapPartitions(func).collect() | ||
return reduce(op, vals, zeroValue) | ||
|
||
# TODO: aggregate | ||
def aggregate(self, zeroValue, seqOp, combOp): | ||
""" | ||
Aggregate the elements of each partition, and then the results for all | ||
the partitions, using a given combine functions and a neutral "zero | ||
value." | ||
|
||
The functions C{op(t1, t2)} is allowed to modify C{t1} and return it | ||
as its result value to avoid object allocation; however, it should not | ||
modify C{t2}. | ||
|
||
The first function (seqOp) can return a different result type, U, than | ||
the type of this RDD. Thus, we need one operation for merging a T into an U | ||
and one operation for merging two U | ||
|
||
>>> seqOp = (lambda x, y: (x[0]+y, x[1] + 1)) | ||
>>> combOp = (lambda x, y: (x[0]+y[0], x[1] + y[1])) | ||
>>> sc.parallelize([1, 2, 3, 4]).aggregate((0, 0), seqOp, combOp) | ||
(10, 4) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. You should add a test where you |
||
""" | ||
def func(iterator): | ||
acc = zeroValue | ||
for obj in iterator: | ||
acc = seqOp(acc, obj) | ||
if acc is not None: | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. acc will never be None, right, because you set it to zeroValue? You can probably just |
||
yield acc | ||
return self.mapPartitions(func).reduce(combOp) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. nit: new line after |
||
def sum(self): | ||
""" | ||
Add up the elements in this RDD. | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
nit: spacing around
+
is inconsistent