0

I have two Dataframe :

  1. A Dataframe DF1 with structure like : (ID, StartDate, EndDate, Position)

  2. a Dataframe DF2 that look like : (DateTime, Position)

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)

We can assume I have a udf function isNearUDF(pos1,pos2) that does the job to comparate positions.

I'm currently trying to do this with a join between my dataframes, but it does not seems to be the right solution

EDIT 2:

Here is a MVCE :

def isInRadius(lat1:Double,lon1:Double,lat2:Double,lon2:Double,dist:Double):Boolean={
  val distance = 0// calculate distance between lon/lat positions

  return distance<=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">=$"CreationDate" && $"DateTime"<=$"EndDate" /*&& isInRadiusUdf($"Lat1",$"Lon1",$"Lat2",$"Lon2",lit(10))*/)

display(DF3)

That work for dates, but take a long time. When I add the isInRadius condition, I get the error :

SparkException: Job aborted due to stage failure: Task not serializable: java.io.NotSerializableException: org.apache.spark.sql.DataFrameReader
6
  • 1
    Why a join isn´t a good solution? Apr 4, 2018 at 9:45
  • I can't manage to use my UDF function while doing a join, I will provide an example of my code. Moreover, the join while checking if the DateTime is in the timerange take a lot of time
    – Nakeuh
    Apr 4, 2018 at 10:02
  • 1
    without joining you won't be able to perform udf function on both dataframes at once. You don't have to perform udf function if you apply the logic while joining Apr 4, 2018 at 10:05
  • provide us the samples of both dataframes with schema and expected output Apr 4, 2018 at 10:20
  • We need a MVCE please !
    – eliasah
    Apr 4, 2018 at 11:49

2 Answers 2

0

Try to change your function definition to:

def isInRadius : Double => Double => Double => Double => Double = lat1 => long1 => lat2 => long2 => dist {
  val distance = // calculate distance between lon/lat positions

  return distance<=dist
}
1
  • What is different with this syntax ? When I try this I get 2 errors : error: not found: value dist def isInRadius : Double => Double => Double => Double => Double = lat1 => long1 => lat2 => long2 => dist { error: not found: value dist return distance<=dist
    – Nakeuh
    Apr 4, 2018 at 11:50
0

After trying various possible solutions and getting weird result, I finally managed to solve my problem by simply restarting my Spark Cluster (Databricks Notebook) I have absolutely no idea what was the problem but the code of the MVCE work now.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

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