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:


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:

  .select(calc_dist_udf('x', 'y', 'x_r', 'y_r').alias('dist'))\

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?

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

  • $\begingroup$ Thanks, I will review that source. Unfortunately I am performing a variable buffer on all geometries (or in this case centroids) in a spatial dataset, and accumulating the smoothed values for each buffer region. There is no ceiling on buffer value, so theoretically a buffer could encompass the entire area being evaluated (which can be composed of upwards of 200k geoms). I understand perf is inherently exponential, but I'd expect Spark, when provisioned w/ sufficient resources, would have the effect of shifting that curve to the right. I have not observed this when dealing with n<=200k geoms. $\endgroup$
    – kuanb
    Sep 15 '18 at 19:36
  • 1
    $\begingroup$ You can manually create a sort of 'broadcast join' by simply loading the whole data set into memory on each executor, and looping over it for every data point. For 200K data points that's not bad to side-load, and should be about as efficient as anything the SQL engine can figure out. There are about 20B pairs to compare though ... the size of the result is huge no matter what. Also consider dropping the sqrt if you just need to compare sizes, not compute them exactly. $\endgroup$
    – Sean Owen
    Sep 16 '18 at 1:25

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


Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.