[SPARK-25299] Yet another attempt to integrate API with scheduler

[yifeih]

No description provided.

[mccheah]

I'll look more thoroughly later, just wanted to get the general philosophy of this approach. This looks sound to me - @squito should give another opinion. This isn't comprehensive - e.g. you can't use this to continuously add new locations for a given block after it's been committed once, but I think it goes a long way to solving the numerous contradictions we had with our other tries at this.

[mccheah]

Can just do shuffleLocationComponents.forEach

[mccheah]

Indentation seems weird

[mccheah]

Let's make thus shuffleLocation an Option as well.

[yifeih]

This is pretty similar to #548 except that we move the port/host logic behind the API. Thanks to @mccheah for the idea!

This gives us a cleaner API, allowing the plugin to define the logic for whether to remove the MapStatus or not based on the implementation and structure of ShuffleLocation and MapShuffleLocations.

We avoid putting the shouldRemoveMapOutputOnLostBlock() logic inside of MapShuffleLocations directly to keep the MapShuffleLocations as an immutable object (instead of something with mutable state, which would further complicate the plugin implementation). Furthermore, placing it in the ShuffleLocationsComponents allows extensibility for cases where the immutable MapShuffleLocations object might not have all the information necessary for whether to remove a MapStatus or not. An example would be in the individual file server case. For example, if you had the following in your MapOutputTracker:

• Map task 1
  ** Block 1: [hostA, hostB]
  ** Block 2: [hostA, hostB]
• Map task 2
  ** Block 1: [hostA, hostC]
  ** Block 2: [hostA, hostC]

If we get a FetchFailed([hostA, hostB]), then shouldRemoveMapOutputOnLostBlock() wouldn't remove the MapStatus associated with map task 2 because both blocks can be retrieved from hostC. Suppose that the scheduler gets a FetchFailure([hostC, hostD]) later. It also doesn't mark map task 2's MapStatus as invalid because the MapShuffleLocations object itself doesn't know that hostA has already been invalidated. Therefore, the MapShuffleLocations object itself doesn't have enough state to know whether a certain location should result in deleting that MapStatus entry.

By moving the logic inside of a separate entity like ShuffleLocationComponents, we make it possible for that object to maintain bookkeeping about which hosts have been blacklisted, allowing it to do additional filters for whether the MapStatus should be removed.

Furthermore, we were hesitant about placing ShuffleLocationComponents inside of ShuffleDriverComponents because of the order of initializations within spark. Currently, the SparkContext first initializes the SparkEnv and then calls ShuffleDriverComponents#initializeApplication(). This makes sense because in some implementations, such as the K8s individual file server implementation, it might need to call to retrieve modules from SparkEnv (https://github.com/mccheah/spark/pull/6/files#diff-1a614b54924c44438c4301263685e461R42). However, the MapOutputTracker is initialized inside the SparkEnv, and we'd ideally want to pass the map status removal logic to the MapOutputTracker. If we placed the map status removal logic inside of ShuffleDriverComponents, we would either need to mandate that SparkEnv is never used inside of ShuffleDriverComponents and initialize it before SparkEnv, or keep it initialized afterwards and pass it to the MapOutputTracker afterwards through a MapOutputTracker.initialize() method.

[mccheah]

Think @ifilonenko will want to look at this too

[squito]

I think I understand what's going on with removeMapAtLocation, but I'm confused by this part. When there is a fetch failure from host X for map M from reduce R, the dagscheduler removes shuffle outputs two different ways (no idea why) -- (1) it uses the path you're changing below in removeMapAtLocation to remove all map output on host X. I see how you're changing that part below to move logic into the ShuffleLocationComponent. (2) here, it is removing the Map M from host X. before this change, it removes the entire map output of Map M. You're changing it to take in the reduce R as a parameter, but still removing all map output of Map M (mapStatus(mapId) = null), which doesn't seem consistent.

Or is there some special logic in the extra condition mapStatuses(mapId).mapShuffleLocations.getLocationForBlock(reduceId) == shuffleLoc? I guess I don't understand when it would be true or false -- seems like it should always be true. Can one ShuffleLocation really represent multiple locations? I'd still expect them to be equal here, as the mapOutputTracker would have stored multiple locations, and the fetch failure would also send back multiple locations, right?

(above discussion ignores the host / executor distinction for external shuffle service in current implementation just to keep things "simple")

[yifeih]

Oh the first point about this function taking a reducer parameter R, it's using R to validate that the ShuffleLocation does indeed exist in the MapStatus for mapper M. I was doing this to mimic the previous behavior where you're checking that the bmAddress exists in the MapStatus before removing it, although I wasn't exactly sure why one would need to check (could it be that we might have received an obsolete FetchFailed?). However, you're right that I can replace the mapStatuses(mapId).mapShuffleLocations.getLocationForBlock(reduceId) == shuffleLoc logic with something that calls into the logic inside ShuffleLocationComponent, which would be more consistent.

On the note about ShuffleLocation representing multiple locations: yup that's right, I tried to code up what that would look like in DAGSchedulerFileServerSuite. Fetch failure would then send back a single ShuffleLocation that encodes multiple host/port combos.

[ifilonenko]

Can an obsolete FetchFailed be a problem, I would leave the check

Does order in the multiple host/port combos matter in this equality check?

[squito]

i think you mean "This removes the map output from the first task, but the map output from the second task is still intact because it was uploaded to the remote resilient datastore".

But -- how do you know that it was uploaded successfully? The executor might have been lost before that upload succeeded too, right?

[yifeih]

Yea you're right, it definitely could have been lost too, but it's assuming the best case scenario right now... I'm wondering whether there should be logic inside of ShuffleLocationComponents to ask the dfs whether the other MapStatus blocks exist or not. Otherwise, the driver is completely in the blind when it comes to answering that question, and it would either 1) assume best case scenario (that the blocks do exist) and sacrifice an attemptId increment to retrying the first mapper that definitely did fail, or 2) assume worst case scenario and recalculate everything on that executor, which might do a lot more work than necessary. Neither 1 or 2 seem ideal

[squito]

I think you pointed out the problem with (1) on another PR or doc something -- if its really gone, then you spend a lot of time retrying to discover all of the blocks which have been lost, and it also might put you over the retry limit for just one failure. This isn't that unlikely a scenario -- we're already talking about dealing with some sort of failure here, doesn't seem unreasonable that the node was slow to do the async backups before that (eg. because it was overloaded or has some hardware which makes it limp along etc.)

[ifilonenko]

In the async to dfs backup case I thought that there can be no assumption by the shuffle fetcher that the backup write was successful. And the best way to know if it exists is by doing simultaneous fetches on all host:port combinations in the shuffle location until one returns a result. I dont think it is necessary for the driver to have a heartbeat check on whether the dfs contains the MapStatus block. I mean, the usual scenario would be that you would read from local and only use the dfs as backup. I guess w.r.t. to the cases you listed, for no blocks to exist this would mean that your local blocks are missing and it failed to backup (if local is missing it probably wasn't immediate... maybe there is a cleanup script run after backup completes or the node fails after a long time) so I think you can assume there is a catastrophic failure on the DFS side if the backup is missing the block, which I would assume would kick off case 2.

I agree, that in reference to @squito question, it is safe to say that this is testing the case that the write was successful.

[squito]

not sure what this is supposed to be doing -- the method's return type is Unit so this is strange. is this supposed to be fail(s"missing map status for $shuffleId")? Or should the map above be a flatMap and this gets ignored?

[yifeih]

It's mapping all the mapStatuses to their corresponding mapShuffleLocations, but mapStatus could be null since the shuffle writer plugin doesn't necessarily have to return a MapShuffleLocation upon completion of the write. It maps the nulls to null so I can assert that the nulls exist in the mapOutputTracker.

[squito]

I'm still confused. don't you see a warning msg from the compiler like

warning: enclosing method assertMapOutputTrackerContains has result type Unit: return value discarded
             return null

?

if you've got a mapStatuses of Array(someMapStatus, null, anotherMapStatus), what do you want this method to do? Is it supposed to skip the assert below entirely? (I think that is what it is doing). Do you want it to check the non-null members (eg. so you'd only expect it to pass if set had two elements, not three?

Or do you want it to check the mapShuffleLocations, unless its null, in which case keep it null? Eg. the check would pass if set == Array(someShuffleLocations, null, anotherShuffleLocations). If its this, then I think you want

map { mapStatus =>
  if (mapStatus == null) {
    null
  } else {
    mapStatus.mapShuffleLocations
  }
}

or even map { mapStatus => Option(mapStatus).map(_.mapShuffleLocations).getOrElse(null)}

[squito]

general comment -- one thing these added tests are missing are cases involving multiple shuffles. In particular, this is important for understanding what you do with shuffle data that is not involved in the current shuffle, eg. rddA --> (groupBy userId) --> rddB ---> (groupBy Date) --> rddC --> (result). If you get a fetch failure when you're trying to compute rddC, what do you assume about the shuffle output generated from the first shuffle on the same host?

I don't think these tests are actually really going through the removeMapAtLocations code you added because they are only touching data of the active shuffle (or maybe it is going through the code, but not in a very interesting way because the data has already been removed)

[yifeih]

I think they are going through the removeMapAtLocations codepath because that's the only way you're going to remove other mapStatuses (mapOutputTracker.unregisterMapOutput() only removes the mapId that failed, and the third test here asserts that two MapStatuses are removed from a single fetch failure). However, you are right that the test cases don't deal with multiple shuffleIds.

[squito]

Want to make sure I understand what is going on here correctly, especially with storing each (map, reduce) shuffle block in its own location, as opposed to just storing the entire map output together. When these tasks re-run, their going to regenerate the entire map output, though actually only one reduce is missing. So now the shuffle server needs to deal w/ getting another version of the output for the (map, reduce) block which was fine before. I think to be consistent with the other ways spark handles non-deterministic output (eg. from a repartition()), it needs to take all of the latest output and throw away whatever it had before.

Does that sound right? I guess its OK as long as its clearly documented, but does seem like an extra headache for implementations.

[yifeih]

Yea, that's how we've been thinking about it too. It's true that when a single reduce is missing, the entire map output will need to be recomputed.

Because of this "overwriting" of still-existing data, we were talking about adding an extra "attemptId" to the API (http://apache-spark-developers-list.1001551.n3.nabble.com/DISCUSS-SPARK-25299-SPIP-Shuffle-storage-API-td27209.html) to ensure that we don't read half-overwritten data.

Also, i'm not sure why it would be extra headache for the implementations? Doesn't the scheduler handle invalidating the already-computed data already in the case of non-deterministic outputs? In the case of deterministic outputs, a successful reducer task that completed before a FetchFailed should still be fine and doesn't need to be recomputed, and the scheduler handles that right now right?

[squito]

yeah I guess I shouldn't say its a headache for implementations, that is an exaggeration -- it just seems like an extra thing to keep straight and a bit more confusing, to me, anyway.

[squito]

super-nit: the behavior would be a little more obvious, to me anyway, if this took two tuples as arguments instead makeFileServerMapStatus(("hostA", "hostB"), ("hostC", "hostD")). On first read I thought there were 4 replicas and I was quite confused.

[ifilonenko]

Why not .foreach() here as well?