title: Pyspark nested loop in the same DataFrame. How to iterate?
body: <p><strong>EDIT</strong>. <strong>TL;DR:</strong>
  I'm trying to achieve a nested loop in a pyspark Dataframe. As you may see,I want the nested loop to start from the NEXT row (in respect to the first loop) in every iteration, so as to reduce unneccesary iterations. Using Python , I can use <strong>[row.Index+1:]</strong> . Generally, in plain Python I can achieve that with the next code:</p>
  
  <pre><code>r=1000
  joins=[]
  for row in df.itertuples():
      for k in df[row.Index+1:].itertuples():
          a = np.array((row.x, row.y))
          b = np.array((k.x, k.y))
          if row.class!= k.class:
              if row.x &lt; k.x+r:
                  if np.linalg.norm(a-b) &lt; r:
                      joins.append((row.name, k.name))
  print(joins)
  </code></pre>
  
  <p>How I can achieve the same result in Spark?</p>
  
  <p><strong>Long problem analysis</strong>:
  I have a list </p>
  
  <pre><code>people=[('john', 35, 54, 'A'), ('george', 94, 84, 'B'), ('nicolas', 7, 9, 'B'), ('tom', 86, 93, 'A'), ('jason', 62, 73, 'B'), ('bill', 15, 58, 'A'), ('william', 9, 3, 'A'), ('brad', 73, 37, 'B'), ('cosmo', 52, 67, 'B'), ('jerry', 73, 30, 'A')]
  </code></pre>
  
  <p>and I convert it to a spark Dataframe which I sort it by 'y', descending:</p>
  
  <pre><code>schema = StructType([
      StructField('name', StringType(), True),
      StructField('x', IntegerType(), True),
      StructField('y', IntegerType(), True),
      StructField('class', StringType(), True),
  ])
  rdd = spark.sparkContext.parallelize(people)
  df = spark.createDataFrame(rdd,schema)
  df.orderBy('y',ascending=False).show()
  </code></pre>
  
  <pre><code>+-------+---+---+-----+
  |   name|  x|  y|class|
  +-------+---+---+-----+
  |    tom| 86| 93|    A|
  | george| 94| 84|    B|
  |  jason| 62| 73|    B|
  |  cosmo| 52| 67|    B|
  |   bill| 15| 58|    A|
  |   john| 35| 54|    A|
  |   brad| 73| 37|    B|
  |  jerry| 73| 30|    A|
  |nicolas|  7|  9|    B|
  |william|  9|  3|    A|
  +-------+---+---+-----+
  </code></pre>
  
  <p>What I 'm looking for is to Join 'class A' names with 'class B' names, when a condition is fulfilled. Let's say for example , when Euclidean distance (A item - B item) &lt; 10.
  I believe that cross join is not the best idea because it could be time consuming with large datasets. </p>
  
  <p>I think it would be nice if I could iterate.
  pseudocode:</p>
  
  <pre><code>start with row1
  if row1[class] != row2[class]:
      if row1['y'] - row2['y'] &lt; 10:
          if Euclidean distance(row1.item - row2.item) &lt; 10:
              join row1.name, row2.name
          end
      else break
  
  it keeps iterating until row1['y'] - row['y'] &gt;= 10
  
  then a new iteration starts from row2
  if row2[class] != row3[class]:
      etc etc
  </code></pre>
  
  <p>In that way, there will be fewer checks between different rows.
  Which is opposite to cross join where everything is checked against everything resulting in larger execution times.
  But how can this be done in spark?
  Any ideas?</p>
  
  <p>edit. Required output (in any form):</p>
  
  <pre><code>brad, jerry
  </code></pre>
  
  <p>(The only pair that its items belong to different classes &amp; their euclidean distance is 7, which is lower than 10)</p>