title: Spark/PySpark collect_set with a binary column
body: <p>Some test data, with two columns: the first binary (using alphanumeric bytes in this example), the second an integer:</p>
  
  <pre><code>from pyspark.sql.types import *
  from pyspark.sql import functions as F
  
  df = spark.createDataFrame([
      (bytearray(b'0001'), 1),
      (bytearray(b'0001'), 1),
      (bytearray(b'0001'), 2),
      (bytearray(b'0002'), 2)
  ],
  schema=StructType([
      StructField("bin", BinaryType()),
      StructField("number", IntegerType())
  ]))
  </code></pre>
  
  <p>Using collect_set to group by the integer column and then remove duplicates doesn't work, because byte arrays don't support hashing. Hence:</p>
  
  <pre><code>(
      df
      .groupBy('number')
      .agg(F.collect_set("bin").alias('bin_array'))
      .show()
  )
  
  +------+------------+
  |number|   bin_array|
  +------+------------+
  |     1|[0001, 0001]|
  |     2|[0001, 0002]|
  +------+------------+
  </code></pre>
  
  <p>One hacky option is to embed the binary array in a struct and then unwrap them all again afterwards, but I suspect this will result in a huge number of allocations and be very expensive (haven't actually profiled it though):</p>
  
  <pre><code>def unstruct_array(input):
      return [x.bin for x in input]
  
  unstruct_array_udf = F.udf(unstruct_array, ArrayType(BinaryType()))
  
  (
      df
      .withColumn("bin", F.struct("bin"))
      .groupBy('number')
      .agg(F.collect_set("bin").alias('bin_array'))
      .withColumn('bin_array', unstruct_array_udf('bin_array'))
      .show()
  )
  
  +------+------------+                                                           
  |number|   bin_array|
  +------+------------+
  |     1|      [0001]|
  |     2|[0001, 0002]|
  +------+------------+
  </code></pre>
  
  <p>If I try lots of Google search terms around binary types and Spark, there are various answers that say you should wrap arrays if you need hashing. Suggestions include a custom wrapper or by calling Scala's toSeq that creates a Scala WrappedArray. For instance:</p>
  
  <p><a href="https://stackoverflow.com/questions/30785615/reducebykey-with-a-byte-array-as-the-key">ReduceByKey with a byte array as the key</a></p>
  
  <p><a href="https://stackoverflow.com/questions/39754445/how-to-use-byte-array-as-key-in-rdd">How to use byte array as key in RDD?</a></p>
  
  <p>So, options include:</p>
  
  <ol>
  <li>Mapping the underlying RDD to make the binary field a WrappedArray. Not sure how to do that in Python?</li>
  <li>Creating a Python wrapper for the array, and then somehow hashing the underlying Java array in Python? Though not sure that has any advantage over using a struct?</li>
  <li>I could wrap in a struct and then never unwrap, which would be a bit more efficient processing-wise, but would then presumably make parquet files bigger and more expensive to parse in all downstream tasks</li>
  </ol>