Wsrtjtyk

Given a dataframe which has the parameters for Http calls to a web accessible API, I want to distribute those API calls to all nodes of a cluster, have make the calls, and return a new dataframe with the result of the calls. This is relatively straightforward, and I have a solution that works.

What is not straightforward is limiting the rate at which the API calls are made. If, for a example, you can only make 100 API calls per second, I want some way to ensure that we don't exceed the limit. Also, if run on a larger cluster, I want some way to ensure that this distributed API caller doesn't turn into a low-grade DDoS attack.

One workable solution is to have each thread sleep when it gets a rejection message from the server (HTTP 429 - too many requests). However, at this point you have already performed the DDoS attack; I'd like to slow down before we get to that point.

I attempted to implement this through an accumulator and broadcast variable. Each call implemented the accumulator and the broadcast variable was the start time. Each worker could then divide the accumulator by time elapsed to see if the request rate was too high. Unfortunately, you can't read an accumulator from a worker. This doesn't work, and I don't see any way to make it work.

I could use the same solution, except control the rate by reading from the driver. I could divide the dataset into a bunch of small partitions, perhaps 10 or 100 each. Then the driver could check the rate before mapping out each partition. However, I don't know any way to introduce a conditional sleep statement executed on the driver side into a .mapPartition() call.

The workflow looks something like this (in Spark 1.6.3)

input.repartition(repartitionSetter(n))

  .select(...fields of interest...)

  .as[(... Tuple DataSet specification ...)]

  .mapPartitions(distributedApiCalls)

  .toDF().toDF( ... column name specification ...)

The conditional statement would work like this:

while (tooManyCalls()) {

  logger.log("Throttling API rate in APPNAME")

  Thread.sleep(1000)

}



def tooManyCalls(): Boolean = {

  val now = Calendar.getInstance.getTimeinMillis

  val timeElapsed = (now - broadcastStartTime.value) / 1000

  (accumulator.value + n) > (timeElapsed * rateLimitPerSec) // true if going too fast

}

Here, repartitionSetter would divide the dataset into partitions of size n and distributedAPICalls is the function passed to each partition to access the API.

Is there any way to incorporate a conditional statement into the distributed API call workflow before MapPartion?

We are in the process of an upgrade to Spark 2.X, but this feature is due in before that upgrade, so an ideal solution will work for both Spark 1.6.3 and Spark 2.X