CODEWITHSUNDEEP
apache-sparkpysparkapache-spark-sql
André Cruz
André Cruz

Reputation: 510

How to sum distances between data points in a dataset using (Py)Spark?

I have a dataset of locations in Lat/Lon format of users in a time period. I would like to calculate the distance these users traveled. Sample dataset:

| Timestamp| User| Latitude|Longitude| |1462838468|49B4361512443A4DA...|39.777982|-7.054599| |1462838512|49B4361512443A4DA...|39.777982|-7.054599| |1462838389|49B4361512443A4DA...|39.777982|-7.054599| |1462838497|49B4361512443A4DA...|39.777982|-7.054599| |1465975885|6E9E0581E2A032FD8...|37.118362|-8.205041| |1457723815|405C238E25FE0B9E7...|37.177322|-7.426781| |1457897289|405C238E25FE0B9E7...|37.177922|-7.447443| |1457899229|405C238E25FE0B9E7...|37.177922|-7.447443| |1457972626|405C238E25FE0B9E7...| 37.18059| -7.46128| |1458062553|405C238E25FE0B9E7...|37.177322|-7.426781| |1458241825|405C238E25FE0B9E7...|37.178172|-7.444512| |1458244457|405C238E25FE0B9E7...|37.178172|-7.444512| |1458412513|405C238E25FE0B9E7...|37.177322|-7.426781| |1458412292|405C238E25FE0B9E7...|37.177322|-7.426781| |1465197963|6E9E0581E2A032FD8...|37.118362|-8.205041| |1465202192|6E9E0581E2A032FD8...|37.118362|-8.205041| |1465923817|6E9E0581E2A032FD8...|37.118362|-8.205041| |1465923766|6E9E0581E2A032FD8...|37.118362|-8.205041| |1465923748|6E9E0581E2A032FD8...|37.118362|-8.205041| |1465923922|6E9E0581E2A032FD8...|37.118362|-8.205041|

I have thought of using a custom aggregator function but it seems there is no Python support for this. Moreover the operations need to be done on adjacent points in a specific order, so I don't know if a custom aggregator would work.

I have also looked at reduceByKey but the operator requirements don't seem to be met by the distance function.

Is there a way to perform this operation in an efficient manner in Spark?

Upvotes: 1

Views: 5934

Answers (1)

zero323
zero323

Reputation: 330193

It looks like a job for window functions. Assuming we define distance as:

from pyspark.sql.functions import acos, cos, sin, lit, toRadians

def dist(long_x, lat_x, long_y, lat_y):
    return acos(
        sin(toRadians(lat_x)) * sin(toRadians(lat_y)) + 
        cos(toRadians(lat_x)) * cos(toRadians(lat_y)) * 
            cos(toRadians(long_x) - toRadians(long_y))
    ) * lit(6371.0)

you can define window as:

from pyspark.sql.window import Window

w = Window().partitionBy("User").orderBy("Timestamp")

and compute distances between consecutive observations using lag:

from pyspark.sql.functions import lag

df.withColumn("dist", dist(
    "longitude", "latitude",
    lag("longitude", 1).over(w), lag("latitude", 1).over(w)
).alias("dist"))

After that you can perform standard aggregation.

Upvotes: 16

Related Questions