The initializer is a tuple (0, 0) it seems you just have 0 From: "subscripti...@prismalytics.io<mailto:subscripti...@prismalytics.io>" Organization: PRISMALYTICS, LLC. Reply-To: "subscripti...@prismalytics.io<mailto:subscripti...@prismalytics.io>" Date: Tuesday, April 28, 2015 at 1:28 PM To: Silvio Fiorito, Todd Nist Cc: "user@spark.apache.org<mailto:user@spark.apache.org>" Subject: Re: Calculating the averages for each KEY in a Pairwise (K,V) RDD ...
Thank you Todd, Silvio...
I had to stare at Silvio's answer for a while.
If I'm interpreting the aggregateByKey() statement correctly ...
(Within-Partition Reduction Step)
a: is a TUPLE that holds: (runningSum, runningCount).
b: is a SCALAR that holds the next Value
(Cross-Partition Reduction Step)
a: is a TUPLE that holds: (runningSum, runningCount).
b: is a TUPLE that holds: (nextPartitionsSum, nextPartitionsCount).
Under that interpretation, I tried to write & run the Python equivalent, like
so:
rdd1.aggregateByKey(0, lambda a,b: (a[0] + b, a[1] + 1), lambda a,b:
(a[0] + b[0], a[1] + b[1]))
Sadly, it didn't work, yielding the following exception which indicates that
the indexing above is incorrect:
lambda v: seqFunc(createZero(), v), seqFunc, combFunc, numPartitions)
File "<input>", line 1, in <lambda>
TypeError: 'int' object has no attribute '__getitem__'
Sidenote: Surprisingly, there isn't much documentation -- at least not for
Python -- for this useful aggregateByKey()
method and use case; although I will be sure to write a gist today, once I get
this working. :)
I think I'm nearly there though, so...
(1) Is my written interpretation above about of what (a,b) are correct?
(2) If yes, what then, is getting passed in the Python case?
I guess I'm looking for the Python equivalent to the first statement in
Silvio's answer (below). But my
reasoning to deconstruct and reconstruct is missing something.
Thanks again!
On 04/28/2015 11:26 AM, Silvio Fiorito wrote:
If you need to keep the keys, you can use aggregateByKey to calculate an avg of
the values:
val step1 = data.aggregateByKey((0.0, 0))((a, b) => (a._1 + b, a._2 + 1), (a,
b) => (a._1 + b._1, a._2 + b._2))
val avgByKey = step1.mapValues(i => i._1/i._2)
Essentially, what this is doing is passing an initializer for sum and count,
then summing each pair of values and counting the number of values. The last
argument is to combine the results of each partition, if the data was spread
across partitions. The result is a tuple of sum and count for each key.
Use mapValues to keep your partitioning by keys intact and minimize a full
shuffle for downstream keyed operations. It just calculates the avg for each
key.
From: Todd Nist
Date: Tuesday, April 28, 2015 at 10:20 AM
To: "subscripti...@prismalytics.io<mailto:subscripti...@prismalytics.io>"
Cc: "user@spark.apache.org<mailto:user@spark.apache.org>"
Subject: Re: Calculating the averages for each KEY in a Pairwise (K,V) RDD ...
Can you simply apply the
https://spark.apache.org/docs/1.3.1/api/scala/index.html#org.apache.spark.util.StatCounter
to this? You should be able to do something like this:
val stats = RDD.map(x => x._2).stats()
-Todd
On Tue, Apr 28, 2015 at 10:00 AM,
subscripti...@prismalytics.io<mailto:subscripti...@prismalytics.io>
<subscripti...@prismalytics.io<mailto:subscripti...@prismalytics.io>> wrote:
Hello Friends:
I generated a Pair RDD with K/V pairs, like so:
>>>
>>> rdd1.take(10) # Show a small sample.
[(u'2013-10-09', 7.60117302052786),
(u'2013-10-10', 9.322709163346612),
(u'2013-10-10', 28.264462809917358),
(u'2013-10-07', 9.664429530201343),
(u'2013-10-07', 12.461538461538463),
(u'2013-10-09', 20.76923076923077),
(u'2013-10-08', 11.842105263157894),
(u'2013-10-13', 32.32514177693762),
(u'2013-10-13', 26.249999999999996),
(u'2013-10-13', 10.693069306930692)]
Now from the above RDD, I would like to calculate an average of the VALUES for
each KEY.
I can do so as shown here, which does work:
>>> countsByKey = sc.broadcast(rdd1.countByKey()) # SAMPLE OUTPUT of
>>> countsByKey.value: {u'2013-09-09': 215, u'2013-09-08': 69, ... snip ...}
>>> rdd1 = rdd1.reduceByKey(operator.add) # Calculate the numerator (i.e. the
>>> SUM).
>>> rdd1 = rdd1.map(lambda x: (x[0], x[1]/countsByKey.value[x[0]])) # Divide
>>> each SUM by it's denominator (i.e. COUNT)
>>> print(rdd1.collect())
[(u'2013-10-09', 11.235365503035176),
(u'2013-10-07', 23.39500642456595),
... snip ...
]
But I wonder if the above semantics/approach is the optimal one; or whether
perhaps there is a single API call
that handles common use case.
Improvement thoughts welcome. =:)
Thank you,
nmv
--
Sincerely yours,
Team PRISMALYTICS
________________________________
PRISMALYTICS, LLC.<http://www.prismalytics.com/>|
www.prismalytics.com<http://www.prismalytics.com/>
P: 212.882.1276<tel:212.882.1276> |
subscripti...@prismalytics.io<mailto:subscripti...@prismalytics.io>
[Follow Us:
https://www.LinkedIn.com/company/prismalytics]<https://www.linkedin.com/company/prismalytics>
[Prismalytics, LLC.]<http://www.prismalytics.com/>
data analytics to literally count on
Title: 403 - Forbidden Error
403 - Forbidden Error
You are not allowed to access this address.
If the error persists, please contact the website webmaster.
If you are the webmaster of this site please log in to Cpanel and check the Error Logs. You will find the exact reason for this error there.
Common reasons for this error are:
- Incorrect file/directory permissions: Below 644.
In order files to be read by the webserver, their permissions have to be equal or above 644. You can update file permissions with a FTP client or through cPanel's File Manager.
- Restrictive Apache directives inside .htaccess file.
There are two Apache directives which can cause this error - 'Deny from' and 'Options -Indexes'.
--------------------------------------------------------------------- To unsubscribe, e-mail: user-unsubscr...@spark.apache.org For additional commands, e-mail: user-h...@spark.apache.org
