Prepare my bigdata with Spark via Python
<p>My 100m in size, quantized data:</p>

<pre><code>(1424411938', [3885, 7898])
(3333333333', [3885, 7898])
</code></pre>

<p>Desired result:</p>

<pre><code>(3885, [3333333333, 1424411938])
(7898, [3333333333, 1424411938])
</code></pre>

<p>So what I want, is to transform the data so that I group 3885 (for example) with all the <code>data[0]</code> that have it). Here is what I did in <a href="/questions/tagged/python" class="post-tag" title="show questions tagged &#39;python&#39;" rel="tag">python</a>:</p>

<pre><code>def prepare(data):
    result = []
    for point_id, cluster in data:
        for index, c in enumerate(cluster):
            found = 0
            for res in result:
                if c == res[0]:
                    found = 1
            if(found == 0):
                result.append((c, []))
            for res in result:
                if c == res[0]:
                    res[1].append(point_id)
    return result
</code></pre>

<p>but when I <code>mapPartitions()</code>'ed <code>data</code> RDD with <code>prepare()</code>, it seem to do what I want only in the current partition, thus return a bigger result than the desired.</p>

<p>For example, if the 1st record in the start was in the 1st partition and the 2nd in the 2nd, then I would get as a result:</p>

<pre><code>(3885, [3333333333])
(7898, [3333333333])
(3885, [1424411938])
(7898, [1424411938])
</code></pre>

<p>How to modify my <code>prepare()</code> to get the desired effect? Alternatively, how to process the result that <code>prepare()</code> produces, so that I can get the desired result?</p>

<hr>

<p>As you may already have noticed from the code, I do not care about speed at all.</p>

<p>Here is a way to create the data:</p>

<pre><code>data = []
from random import randint
for i in xrange(0, 10):
    data.append((randint(0, 100000000), (randint(0, 16000), randint(0, 16000))))
data = sc.parallelize(data)
</code></pre>

<p>You can use a bunch of basic pyspark transformations to achieve this.</p>

<pre><code>&gt;&gt;&gt; rdd = sc.parallelize([(1424411938, [3885, 7898]),(3333333333, [3885, 7898])])
&gt;&gt;&gt; r = rdd.flatMap(lambda x: ((a,x[0]) for a in x[1]))
</code></pre>

<p>We used <code>flatMap</code> to have a key, value pair for every item in <code>x[1]</code> and we changed the data line format to <code>(a, x[0])</code>, the <code>a</code> here is every item in <code>x[1]</code>. To understand <code>flatMap</code> better you can look to the documentation.</p>

<pre><code>&gt;&gt;&gt; r2 = r.groupByKey().map(lambda x: (x[0],tuple(x[1])))
</code></pre>

<p>We just grouped all key, value pairs by their keys and used tuple function to convert iterable to tuple.</p>

<pre><code>&gt;&gt;&gt; r2.collect()
[(3885, (1424411938, 3333333333)), (7898, (1424411938, 3333333333))]
</code></pre>

<p>As you said you can use [:150] to have first 150 elements, I guess this would be proper usage:</p>

<p><code>r2 = r.groupByKey().map(lambda x: (x[0],tuple(x[1])[:150]))</code></p>

<p>I tried to be as explanatory as possible. I hope this helps.</p>

python algorithm apache-spark bigdata distributed-computing