Q:

Combinations And Mass Set Intersections With Apache Spark (Pyspark)

The Challenge:
The data is an RDD in the shape of [(u, p), (u, p), ...] where u and p are both strings. My challenging desired output is [((p,p), u), ..] where the key in each element is a combination of all p's that share a u.
Already Tried:

Using cartesian and filter to simulate combination.
Tried to use itertools.combinations in Python for each element and then flatMap, like something like this:[(tuple(sorted(e)), x[0]) for e in combinations(x[1].split(','),2)])

The problem with what I already tried is that the executor nodes seem to go down, presumably from memory consumption. 
Any suggestions?
EDIT
Here's some more info and background for the question:
I'm looking to find the overlap count/cardinality of the intersection for every two "p"s where a "p" is a collection id and "u" is a member of a collection. 
The input is a huge list of records of relationships between a "p" and a "u", which can be for example: 
[(u1,p1), (u2, p2), (u1, p2), (u2, p1), (u1, p3)]
The desired output is: [((p1,p2), 2), ((p1, p3), 1), ((p2, p3), 1)] 
(note combinations and not permutations)
A few more details about the input data:

There are 50k - 100k different "p"s.
A "p" can have a few hundred million unique "u"s.
There can be duplicate rows of input data. In other words multiple instances of (u,p) but that doesn't change the problem when it comes to treating the list of "u"s as a set for a given "p" for the intersection step.

As for hardware, I'm using about 41 m3.xlarges on EMR (Spark 1.3.1 on Yarn) and kicking the shell off this way:
./pyspark --master yarn-client --driver-memory 4G --executor-memory 3G --num-executors 160

A:

yourRdd.groupByKey().map(lambda (a,b): (b,a))

The group by key will aggregate all the values of "u" into
[(u1,(p1,p2,p3..,px)),(ux,(p1,...,px))]

The map will swap the keys and the values, this should give the desired output.  groupByKey spills to disk when not enough memory is available, so this should alleviate your potential memory consumption issue as well.
//EDIT based on the new information.
I can't provide an exact answer, but I can help walk you through a way that should work with most of the code:
Once you do the groupByKey, the data is partitioned such that every instance of a single key is on the same machine.  Now that we know all the "p's" for a single "u" are on the same machine, we can start manipulating the values.
Lets take my initial code as a starting point, but modify it a bit.
rdd_1 = yourRdd.groupByKey()
rdd_2 = rdd1.mapValues(mapFunction).flatMapValues()

The mapFunction is where the magic will happen.  The function will take a tuple input of all the values of p, and then output a list of tuples where each tuple is a pair of p values.
#Updated based on Zero's recommendation of generators.
def mapFunction(tple):
    l = list(tple)
    for i in range(len(tple))
        for j in l[i+1:]
             yield (l[i], j)

You'll need to add something to handle when you get to the i=len(tple) so we don't an out of bounds exception.  You still might run into memory issues, but the next part should help you on that.
I think we can mess with your executor configurations.  Given your machine set up, I think we can create much LARGER executors.  In addition your application isn't caching data that I'm aware of, so we can set all of the executor memory to be used for objects my modifying the memoryFraction setting.  I've found a few bigger executors are better than a many smaller executors  (though running with YARN, its tougher to get large amount of resources all the time, but thats a whole different conversation).  Try to use executors between 16 and 32gb, with 2-5 cores.
I would do a spark sumbit similar to this:
spark-submit --master yarn-client --driver-memory 4g --executor-memory 16g --num-executors 30 --executor-cores 4  --conf spark.storage.memoryFraction=0

Skip the memoryFraction setting if you are caching/persisting data anywhere

