Pyspark RDD ReduceByKey Multiple function

Question

I have a PySpark DataFrame named DF with (K,V) pairs. I would like to apply multiple functions with ReduceByKey. For example, I have following three simple functions:

def sumFunc(a,b): return a+b

def maxFunc(a,b): return max(a,b)

def minFunc(a,b): return min(a,b)

When I apply only one function, e.g,, following three work:

DF.reduceByKey(sumFunc)  #works
DF.reduceByKey(maxFunc)  #works
DF.reduceByKey(minFunc)  #works

But, when I apply more than one function, it does not work, e.g., followings do not work.

DF.reduceByKey(sumFunc, maxfunc, minFunc) #it does not work
DF.reduceByKey(sumFunc, maxfunc) #it does not work
DF.reduceByKey(maxfunc, minFunc) #it does not work
DF.reduceByKey(sumFunc, minFunc) #it does not work

I do not want to use groupByKey because it slows down the computation.

zero323 · Accepted Answer · 2016-02-23 19:11:13Z

1

If input is a DataFrame just use agg:

import pyspark.sql.functions as sqlf

df = sc.parallelize([
   ("foo", 1.0), ("foo", 2.5), ("bar", -1.0), ("bar", 99.0)
]).toDF(["k", "v"])

df.groupBy("k").agg(sqlf.min("v"), sqlf.max("v"), sqlf.sum("v")).show()

## +---+------+------+------+
## |  k|min(v)|max(v)|sum(v)|
## +---+------+------+------+
## |bar|  -1.0|  99.0|  98.0|
## |foo|   1.0|   2.5|   3.5|
## +---+------+------+------+

With RDDs you can use statcounter:

from pyspark.statcounter import StatCounter

rdd = df.rdd
stats = rdd.aggregateByKey(
    StatCounter(), StatCounter.merge, StatCounter.mergeStats
).mapValues(lambda s: (s.min(), s.max(), s.sum()))

stats.collect()
## [('bar', (-1.0, 99.0, 98.0)), ('foo', (1.0, 2.5, 3.5))]

Using your functions you could do something like this:

def apply(x, y, funs=[minFunc, maxFunc, sumFunc]):
    return [f(x_, y_) for f, x_, y_ in zip(*(funs, x, y))]

rdd.combineByKey(lambda x: (x, x, x), apply, apply).collect()
## [('bar', [-1.0, 99.0, 98.0]), ('foo', [1.0, 2.5, 3.5])]

edited Feb 23, 2016 at 19:11

answered Feb 23, 2016 at 18:38

zero323

331k108 gold badges982 silver badges958 bronze badges

Sign up to request clarification or add additional context in comments.

4 Comments

Matthias Over a year ago

Can you also use the RDD method for multiple StatCounter instances at once? For instance if you want min/max for different columns in the same aggregation?

Matthias Over a year ago

I tried to use your StatCounter example. But when I try it on a key-value RDD with string-float, then I get this error: TypeError: unbound method merge() must be called with NoneType instance as first argument (got StatCounter instance instead)

zero323 Over a year ago

@Matthias I don't use Python 2 anymore. If you use 2.x you'll have to use functions which use specific instances you receive in seq-op and merge-op.

Matthias Over a year ago

thanks for the hint. After search for weeks, I finally found a really good explanation here.

Collectives™ on Stack Overflow

Pyspark RDD ReduceByKey Multiple function

1 Answer 1

4 Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

4 Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related