I have an issue with filtering a large dataset, which I believe is because of an inefficient join. What I'm trying to do is the following:
Dataset info contains a lot of user data, identified by a user ID and a timestamp. Dataset filter_dates contains, for every user ID, the most recent timestamp processed.
The processing job then needs to determine, from a large input source, what data in info is new and to be processed.
What I tried doing was the following:
info
.join(
broadcast(filter_dates),
info("userid") === filter_dates("userid"),
"left"
)
.filter('info_date >= 'latest_date || 'latest_date.isNull)
But this is incredibly slow (taking many hours), and in the Spark UI I see it processing an absurd amount of data as Input.
Then, just for fun, I tried it with this:
val startDate = // Calculate the minimum startDate from filter_dates
info
.filter('info_date >= startDate)
And this was blazingly fast (Taking some minutes), but of course this is not optimal because it ends up re-processing some data. This leads me to believe there's something fundamentally wrong with how I joined the datasets.
Does anyone know how I can improve the joins?