Exponential runtime on distance matrix I have an exploratory script running a Databricks notebook that performs a simple arithmetic function (Pythagorean theorem) on all possible pairwise combinations of a list of pairs of floats (akin to coordinates).

The values are generated randomly, like so:

vals = np.random.rand(num_samples, 2)


The list is then converted to 2 RDDs of Rows, like so:

rdd = sc.parallelize(vals)
rows_1 = rdd.map(lambda v: Row(x=float(v[0]), y=float(v[1]), join_val=1))
rows_2 = rdd.map(lambda v: Row(x_r=float(v[0]), y_r=float(v[1]), join_val=1))


Which are then registered as tables:

sqlContext.createDataFrame(rows_1).registerTempTable('sdf_1')
sqlContext.createDataFrame(rows_2).registerTempTable('sdf_2')

sdf_1 = sqlContext.table('sdf_1')
sdf_2 = sqlContext.table('sdf_2')


Each table contains the same content, just different columns names. The two are then joined:

sdf_1.join(sdf_2, sdf_1.join_val==sdf_2.join_val).registerTempTable('sdf_join')
sdf_join = sqlContext.table('sdf_join')


With the tables joined, the following UDF is defined:

def calc_dist(x1, y1, x2, y2):
  return math.sqrt((x1-x2)**2 + (y1-y2)**2)

calc_dist_udf = udf(calc_dist, FloatType())


Finally, the operation is performed on all rows:

sdf_join\
  .select(calc_dist_udf('x', 'y', 'x_r', 'y_r').alias('dist'))\
  .filter('dist&lt;0.05')\
  .count()


This operation completes successfully, but I have noticed that, as num_samples increases, the execution time increases exponentially. I believe I am failing to correctly parallelize the row-wise operation. 


Is this assumption correct?
How can I achieve parallelization on such an operation?
For 2 tables of size $n$ and $m$, join operation complexity is in $O(n.m)$ in worst case. If $n=m$ we are facing a quadratic operation which is why you observe slowness with the increase of tables size. Parallelize a quadratic problem requier at least to increase quadratically the number of slaves, without taking account shuffling, to keep constant execution time with increase of problem size which is not really a feasable option...