Spark: How to "reduceByKey" when the keys are numpy arrays which are not hashable?
<p>I have an RDD of (key,value) elements. The keys are NumPy arrays. NumPy arrays are not hashable, and this causes a problem when I try to do a <code>reduceByKey</code> operation.</p>

<p>Is there a way to supply the Spark context with my manual hash function? Or is there any other way around this problem (other than actually hashing the arrays "offline" and passing to Spark just the hashed key)?</p>

<p>Here is an example:</p>

<pre><code>import numpy as np
from pyspark import SparkContext

sc = SparkContext()

data = np.array([[1,2,3],[4,5,6],[1,2,3],[4,5,6]])
rd = sc.parallelize(data).map(lambda x: (x,np.sum(x))).reduceByKey(lambda x,y: x+y)
rd.collect()
</code></pre>

<p>The error is:</p>

<blockquote>
  <p>An error occurred while calling
  z:org.apache.spark.api.python.PythonRDD.collectAndServe.</p>
  
  <p>...</p>
  
  <p>TypeError: unhashable type: 'numpy.ndarray'</p>
</blockquote>

<p>The simplest solution is to convert it to an object that is hashable. For example:</p>

<pre><code>from operator import add

reduced = sc.parallelize(data).map(
    lambda x: (tuple(x), x.sum())
).reduceByKey(add)
</code></pre>

<p>and convert it back later if needed.</p>

<blockquote>
  <p>Is there a way to supply the Spark context with my manual hash function</p>
</blockquote>

<p>Not a straightforward one. A whole mechanism depend on the fact object implements a  <code>__hash__</code> method and C extensions are cannot be monkey patched. You could try to use dispatching to override <code>pyspark.rdd.portable_hash</code> but I doubt it is worth it even if you consider the cost of conversions. </p>

python numpy apache-spark pyspark