Efficient filtering of a large dataset in Spark based on other dataset

Viewed 119

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?

0 Answers
Related