Exponential runtime on distance matrix

Question

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<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.

1. Is this assumption correct?
2. How can I achieve parallelization on such an operation?

Answer 1

You are comparing every sample to every other sample. Of course the runtime increases as the square of the input size, because the work increases as the square of the input size.

There are more efficient ways to do this, if you must; what about a self-join on 0 columns rather than making two copies of the DF with a dummy col? (I have not checked if this works.). Caching the dataframe, and/or ensuring Spark is doing a broadcast join, could help.

But I find you rarely really want to compute all-pairs distances; you probably want to find points that are close? there are much faster algorithms, even in Spark, for this. Look at DIMSUM for example https://stanford.edu/~rezab/slides/maryland_mllib.pdf

Answer 2

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...