EDIT 2022/02/18: I returned to this problem after a few years, and I believe my new solution below is substantially more performant than the current highest-voted solution.

Suppose I have a DataFrame with a column `partition_id`:

```
n_partitions = 2

df = spark.sparkContext.parallelize([
    [1, 'A'],
    [1, 'B'],
    [2, 'A'],
    [2, 'C']
]).toDF(('partition_id', 'val'))

```

How can I repartition the DataFrame to guarantee that each value of `partition_id` goes to a separate partition, and that there are exactly as many actual partitions as there are distinct values of `partition_id`?

If I do a hash partition, i.e. `df.repartition(n_partitions, 'partition_id')`, that guarantees the right number of partitions, but some partitions may be empty and others may contain multiple values of `partition_id` due to hash collisions.