[PYSPARK] Fix a typo in "fold" function in rdd.py

[aihex]

This will make the “fold” function consistent with the "fold" in rdd.scala and other "aggregate" functions where “acc” goes first. Otherwise, users have to write a lambda function like “lambda x, y: op(y, x)” if they want to use “zeroValue” to change the result type.

[AmplabJenkins]

Can one of the admins verify this patch?

[srowen]

I think that's right that it would be more consistent, in that this is how the Scala function that's ultimately called for RDD.fold looks:

  def foldLeft[B](z: B)(op: (B, A) => B): B = {
    var result = z
    this.seq foreach (x => result = op(result, x))
    result
  }

Elements are passed as the second arguments. The operation is commutative but it matters here since the operation is only supposed to modify the first arg.

Why does this affect your ability to make a differently typed result? You need aggregate for that anyway I thought.

CC @JoshRosen as initial author

[aihex]

@JoshRosen

I just follow the example of NAStatCounter in the book of "Advanced Analysis with Spark". NAStatCounter is supposed to get stats of a dataset where some missing values exist.

Instead of using scala the same as the book, I use python to reimplement that. My code follows.

class NAStatCounter(object):

    def __init__(self):
        self.stats = StatCounter()
        self.missing = 0

    def add(self, x):
        if not x:
            self.missing += 1
        elif isinstance(x, NAStatCounter):
            self.stats.mergeStats(x.stats)
            self.missing += x.missing
        else:
            self.stats.merge(float(str(x)))
        return self

    def __str__(self):
        return 'stats: ' + str(self.stats) + ' missing: ' + str(self.missing)

    def __repr__(self):
        return self.__str__()

Here I make up a dummy dataset and the do the stats.

rdd = sc.parallelize(['1,,2', ',3,1', '1,2,'])
na_stat = rdd.map(lambda x: x.split(','))
z = [NAStatCounter for i in xrange(3)]
op = lambda x, y: map(lambda a: a[0].add(a[1]), zip(x, y))
result = na_stat.fold(z, op)

Then I get the error like "'str' object has no attribute 'add'" because it has op('1', NAStatCounter()) in the "fold" implementation. In the specified lambda function, it becomes '1'.add(NAStatCounter()). However, it's expected to be NAStatCounter().add('1').

As you mentioned, only the first argument can be modified and I guess it should be the provided "zeroValue" and the element are the second argument which is not allowed to be changed.

Intuitively, users specify the "zeroValue" as "x" and elements as "y" in the lambda function. To correct that, I have to change op to

op = lambda x, y: map(lambda a: a[0].add(a[1]), zip(y, x))

That's kind of anti-intuitive and inconsistant with its counterpart in scala just as the example in "Advanced Analysis with Spark".

[JoshRosen]

I agree that swapping the order of the function arguments makes more sense, but at this point I wonder whether it's safe to change this. The old code has been around since Spark 0.7 or something and it's possible that someone could be relying on the existing confusing behavior. Maybe we should just point out this inconsistency in the docs.

[srowen]

This feels like it should be tracked as a JIRA, even if one targeted at version 2+ only to actually change.

[aihex]

Thanks for replying. I guess it might be better to remain it unchanged and point it out at docs. I will be trying to do the modification at the doc.