Input and Output of function in pyspark
<p>I'd like to get some lesson about the input grammar in pyspark.</p>

<p>My platform is below.</p>

<pre><code>Red Hat Enterprise Linux Server release 6.8 (Santiago)
spark version 1.6.2
python 2.6
</code></pre>

<p>I have def defined in module <strong>basic_lib.py</strong> as below.</p>

<pre><code>def selectRowByTimeStamp(x,y):
    if x._1 &gt; y._1:
        return x
    return y
</code></pre>

<p>Below is my snippet of the main code</p>

<pre><code>  df2 = df2.map(lambda x: (x._2, x))
  rdd = df2.reduceByKey(basic_lib.selectRowByTimeStamp)
</code></pre>

<p>Why does above <strong>basic_lib.selectRowByTimeStamp</strong> work without clearly 
specifying the input parameter?</p>

<p>For example , something like below is more easy to understand.</p>

<pre><code>var1 = 1
var2 = 2

rdd = df2.reduceByKey(basic_lib.selectRowByTimeStamp(var1, var2))
</code></pre>

<p>It looks like you're slightly confused about what exactly is the purpose of <code>lambda</code> expressions. In general <code>lambda</code> expression in Python are used to create anonymous, single expression functions. Other than than that, as far as we care here, there are not different than any other function you define. To quote the docs:</p>

<blockquote>
  <p>Small anonymous functions can be created with the lambda keyword. (...) Lambda functions can be used wherever function objects are required. <strong>Semantically, they are just syntactic sugar for a normal function definition.</strong> </p>
</blockquote>

<p>Since lambda functions are not special in Python in general there cannot be special in PySpark (well, they may require some serialization tricks due to their scope but it is only about their scope). No matter if function is defined by <code>lambda</code> or not (or if it is even a function*) Spark applies it in exactly the same way. So when you call:</p>

<pre><code>df2.map(lambda x: (x._2, x))
</code></pre>

<p>lambda expression is simply evaluated and what is received by <code>map</code> is just another function object. It wouldn't be different if you assigned if first:</p>

<pre><code>foo = lambda x: (x._2, x)  # Yup, this is against style guide (PEP 8)
</code></pre>

<p>or created a standalone function:</p>

<pre><code>def bar(x):
    return x._2, x 
</code></pre>

<p>In all three cases function object is functionally pretty much the same:</p>

<pre><code>import dis

dis.dis(foo)
##   1           0 LOAD_FAST                0 (x)
##               3 LOAD_ATTR                0 (_2)
##               6 LOAD_FAST                0 (x)
##               9 BUILD_TUPLE              2
##              12 RETURN_VALUE

dis.dis(bar)
##   2           0 LOAD_FAST                0 (x)
##               3 LOAD_ATTR                0 (_2)
##               6 LOAD_FAST                0 (x)
##               9 BUILD_TUPLE              2
##              12 RETURN_VALUE

dis.dis(lambda x: (x._2, x))
##   1           0 LOAD_FAST                0 (x)
##               3 LOAD_ATTR                0 (_2)
##               6 LOAD_FAST                0 (x)
##               9 BUILD_TUPLE              2
##              12 RETURN_VALUE
</code></pre>

<p>On a side note if input is a <code>DataFrame</code> here it is much more efficient to solve this using Spark SQL. Also it is better to extract <code>rdd</code> before you use <code>map</code> to ensure forward compatibility. Finally <code>Row</code> is just a tuple. </p>

<p>So optimally you could:</p>

<pre><code>df.groupBy("_2").max()
</code></pre>

<p>but if you really want to use RDD API:</p>

<pre><code>df.select("_2", "_1").rdd.reduceByKey(max)
</code></pre>

<hr>

<p>* In practice any callable object will work as long as it accepts given arguments. For example (not that it makes much sense here) you could replace function with an object of a class defined as follows:</p>

<pre><code>class FooBar(object):
    def __call__(self, x):
        return x._2, x

df2.rdd.map(FooBar())
</code></pre>

python apache-spark pyspark