Why can't I reference key in reduce logic?
<p>I want to have logic in my <code>combineByKey</code>/<code>reduceByKey</code>/<code>foldByKey</code> that relies on the key currently being operated on. From what I can tell by the method signatures, the only parameters passed to these methods are the values being combined/reduced/folded.</p>

<p>Using a simple example where I just have an RDD which is <code>(int, int)</code> tuples, the result I want is an rdd keyed by <code>tuple[0]</code> where the value is the <code>int</code> closest to the key.</p>

<p>For example:</p>

<pre><code>(1, 8)
(1, 3)
(1, -1)
(2, 4)
(2, 5)
(2, 2)
(3, 2)
(3, 4)
</code></pre>

<p>Should reduce to:</p>

<pre><code>(1, 3)
(2, 2)
(3, 2)
</code></pre>

<p>Note in comparing <code>(1, 3)</code> and <code>(1, -1)</code> I don't care which one is picked since they are both the same distance. Same for the "3" key.</p>

<p>The way I would imagine doing this would be something along the lines of:</p>

<pre><code>rdd.reduceByKey(lambda key, v1, v2: v1 if abs(key - v1) &lt; abs(key - v2) else v2)
</code></pre>

<p>But the <code>reduce</code> function only takes 2 arguments: two values to be combined. It seems like the easiest method would be to reference the key in my reducer in order to achieve my goal; is this possible?</p>

<p>If I try this I get an error:</p>

<pre><code>rdd = sc.parallelize([(1, 8), (1, 3), (1, -1), (2, 4), (2, 5), (2, 2), (3, 2), (3, 4)])
rdd.reduceByKey(lambda key, v1, v2: v1 if abs(key - v1) &lt; abs(key - v2) else v2).collect()
</code></pre>

<blockquote>
  <p>TypeError: () takes exactly 3 arguments (2 given)</p>
</blockquote>

<p>I'm not really looking for a solution to this example problem. What I'm wondering is if there is a reason the key is not passed to the <code>reduceByKey</code> function? I assume it is some basic principle of the map-reduce philosophy that I am missing.</p>

<hr>

<p>Note I can solve my example by inserting a map step which maps each value to a tuple consisting of the value and the distance from the key:</p>

<pre><code>rdd = sc.parallelize([(1, 8), (1, 3), (1, -1), (2, 4), (2, 5), (2, 2), (3, 2), (3, 4)])
rdd = rdd.map(lambda tup: (tup[0], tuple([tup[1], abs(tup[0] - tup[1])])))
rdd.reduceByKey(lambda v1, v2: v1 if v1[1] &lt; v2[1] else v2).mapValues(lambda x: x[0]).collectAsMap()
</code></pre>

<p>I think there is no strong reason not to pass keys.<br>
however, I feel <code>reduceByKey</code> API was designed for common use case -- compute sum of values for each key.  So far I've never needed keys within the value computation.  But that's just my opinion.</p>

<p>Also the problem that you solved seems to be simple aggregation problem.  <code>min()</code> and <code>groupByKey</code> can find the answer. I know you are not looking for a solution but here is how I'd write.</p>

<pre><code>from pyspark import SparkContext

sc = SparkContext()
rdd = sc.parallelize([(1, 8), (1, 3), (1, -1), (2, 4), (2, 5), (2, 2), (3, 2), (3, 4)])
reduced = rdd.groupByKey().map(lambda (k, v): (k, min(v, key=lambda e:abs(e-k))))
print(reduced.collectAsMap())
</code></pre>

<p>Result</p>

<pre><code>{1: 3, 2: 2, 3: 2}
</code></pre>

python mapreduce pyspark