Find distinct values for each column in an RDD in PySpark
<p>I have an RDD that is both very long (a few billion rows) and decently wide (a few hundred columns). I want to create sets of the unique values in each column (these sets don't need to be parallelized, as they will contain no more than 500 unique values per column).</p>

<p>Here is what I have so far:</p>

<pre><code>data = sc.parallelize([["a", "one", "x"], ["b", "one", "y"], ["a", "two", "x"], ["c", "two", "x"]])
num_columns = len(data.first())
empty_sets = [set() for index in xrange(num_columns)]
d2 = data.aggregate((empty_sets), (lambda a, b: a.add(b)), (lambda x, y: x.union(y)))
</code></pre>

<p>What I am doing here is trying to initate a list of empty sets, one for each column in my RDD. For the first part of the aggregation, I want to iterate row by row through <code>data</code>, adding the value in column <code>n</code> to the <code>n</code>th set in my list of sets. If the value already exists, it doesn't do anything. Then, it performs the <code>union</code> of the sets afterwards so only distinct values are returned across all partitions.</p>

<p>When I try to run this code, I get the following error:</p>

<p><code>AttributeError: 'list' object has no attribute 'add'</code></p>

<p>I believe the issue is that I am not accurately making it clear that I am iterating through the list of sets (<code>empty_sets</code>) and that I am iterating through the columns of each row in <code>data</code>. I believe in <code>(lambda a, b: a.add(b))</code> that <code>a</code> is <code>empty_sets</code> and <code>b</code> is <code>data.first()</code> (the entire row, not a single value). This obviously doesn't work, and isn't my intended aggregation.</p>

<p>How can I iterate through my list of sets, and through each row of my dataframe, to add each value to its corresponding set object?</p>

<p>The desired output would look like:</p>

<p><code>[set(['a', 'b', 'c']), set(['one', 'two']), set(['x', 'y'])]</code></p>

<hr>

<p>P.S I've looked at this example <a href="https://stackoverflow.com/questions/31623897/how-to-efficiently-find-distinct-values-from-each-column-in-spark">here</a>, which is extremely similar to my use case (it's where I got the idea to use <code>aggregate</code> in the first place). However, I find the code very difficult to convert into PySpark, and I'm very unclear what the <code>case</code> and <code>zip</code> code is doing.</p>