title: Spark Scala : Join two Dataframes by near position and time range
body: <p>I have two Dataframe : </p>
  
  <ol>
  <li><p>A Dataframe <code>DF1</code> with structure like :  <code>(ID, StartDate, EndDate, Position)</code></p></li>
  <li><p>a Dataframe <code>DF2</code> that look like : <code>(DateTime, Position)</code></p></li>
  </ol>
  
  <p>I want to use those Dataframes to create a new one, that contains for each DF1(ID), the number of rows in DF2 where DF2(DateTime) is between DF1(StartDate) and DF1(EndDate) and DF2(Position) is near DF1(Position)</p>
  
  <p>We can assume I have a udf function <code>isNearUDF(pos1,pos2)</code> that does the job to comparate positions.</p>
  
  <p>I'm currently trying to do this with a join between my dataframes, but it does not seems to be the right solution</p>
  
  <p>EDIT 2: </p>
  
  <p>Here is a MVCE : </p>
  
  <pre><code>def isInRadius(lat1:Double,lon1:Double,lat2:Double,lon2:Double,dist:Double):Boolean={
    val distance = 0// calculate distance between lon/lat positions
  
    return distance&lt;=dist
  }
  
  val DF1 = sc.parallelize(Array(
    ("ID1", "2018-02-27T13:47:59.416+01:00", "2018-03-01T16:02:00.632+01:00", "25.13297154663", "55.13297154663"),
    ("ID2", "2018-02-25T13:47:59.416+01:00", "2018-02-07T16:02:00.632+01:00", "26.13297154663", "55.13297154663"),
    ("ID3", "2018-02-24T13:47:59.416+01:00", "2018-02-02T16:02:00.632+01:00", "25.13297154663", "55.13297154663")
  // ...
  )).toDF("ID", "CreationDate","EndDate","Lat1","Lon1")
  
  val DF2 = sc.parallelize(Array(
    ("2018-02-27T13:47:59.416+01:00","25.13297154663", "55.13297154663"),
    ("2018-02-27T13:47:59.416+01:00","25.1304663", "54.10663"),
    ("2018-02-27T13:47:59.416+01:00","25.1354663", "55.132904663")
    // ...
  )).toDF("DateTime","Lat2","Lon2")
  
  val isInRadiusUdf = udf(isInRadius _)
  
  val DF3 = DF1.join(DF2,$"DateTime"&gt;=$"CreationDate" &amp;&amp; $"DateTime"&lt;=$"EndDate" /*&amp;&amp; isInRadiusUdf($"Lat1",$"Lon1",$"Lat2",$"Lon2",lit(10))*/)
  
  display(DF3)
  </code></pre>
  
  <p>That work for dates, but take a long time.
  When I add the isInRadius condition, I get the error : </p>
  
  <pre><code>SparkException: Job aborted due to stage failure: Task not serializable: java.io.NotSerializableException: org.apache.spark.sql.DataFrameReader
  </code></pre>