Attaching a draft of the proposed design

The design document covers affinity for Samza with YARN and standalone design. Some sections are TBD. Looking for some feedback.

A few comments after the first glance:

1. What’s the implication of yarn.scheduler.fair.locality.threshold. {node,rack}

   ?

2. Step 2 in the proposed solution: Samza AM ==> Samza Job Coordinator? I felt that this part of logic should be outside the implementation of YARN AM, s.t. it should be easily extended to other environment
3. As for standalone Samza, we need to identify whether the host-affinity design applies to that case or not. I.e. if the standalone Samza starts/stops containers and re-shuffle the tasks among the containers every time, it might not be an applicable case. It would be better if we state the assumption at the beginning that this design is only targeted to address the case where task/partition-to-container assignment has not changed.
4. In the yarn-site.xml configure, shouldn’t the yarn,resource manager.scheduler.class set to org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler ?

Another not-so-relevant comment:

1. Samza-557 seems to be only targeted for stores w/ changelog attached? Can we extend it to OFFSET in the single Kafka stream to support window op’s MessageStore recovery?

1.What's the implication of yarn.scheduler.fair.locality.threshold.{node,rack} ?

If you are referring to Step 1 in the solution, then it is a typo. Sorry about that. I was referring to yarn.schedular.fair.locality-delay-{node,rack}-ms. It indicates the amount of time the scheduler wait before relaxing locality that the given network level.

2. Step 2 in the proposed solution: Samza AM ==> Samza Job Coordinator? I felt that this part of logic should be outside the implementation of YARN AM, s.t. it should be easily extended to other environment

You are right. This should not be specific to the Samza AM. It should be in the Job Coordinator. Earlier, Chris Riccomini suggested the same. I will change this.

3. As for standalone Samza, we need to identify whether the host-affinity design applies to that case or not. I.e. if the standalone Samza starts/stops containers and re-shuffle the tasks among the containers every time, it might not be an applicable case. It would be better if we state the assumption at the beginning that this design is only targeted to address the case where task/partition-to-container assignment has not changed.

Agreed.

4. In the yarn-site.xml configure, shouldn’t the yarn,resource manager.scheduler.class set to org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler ?

Again a typo.

Thanks for the review Yi Pan (Data Infrastructure) !

I have edited the document based on Yi's comments.
There is still not sufficient clarity around how to deal with application-specific retention policy and orphaned store clean-up policy.

Steve Loughran ,
I haven't worked with Slider. But Chris Riccomini suggested perhaps you guys have faced a similar issue.

Does Slider provide any kind of persistence of app data beyond the life of a YARN application? That is, is it possible to persist data between multiple application submissions (say, there is job upgrade) beyond the retention period that can be configured in YARN?
If it does, what kind of garbage collection / clean-up policy does it enforce?

We use state stores in Samza that is currently persisted in YARN. In the proposed plan for host-affinity, we plan to move the stores outside the application's cache in YARN so that they can be persisted for longer periods of time (even after the application has exited) and periodically, garbage collect unused or orphaned state stores. Our current line of thought is to use an auxiliary service on the NMs that will clean-up unused state stores, although its event-driven nature makes it hard to tune for periodic tasks.

1. we persist this to HDFS as some avro-formatted placement history, "Role history"
2. every saved entry is timestamped such that the sort order matches version number; its easiest to determine the latest
3. reload is best-effort from the latest-onwards; helps address the "AM crashed partway through a write" problem.
4. after a successful reload, we just delete the old copies.
5. as its all in the filesystem, testing is straightforward

Look at SLIDER-611; I've just checked in the SLIDER-799 patch in which the AM takes over the task of deciding when to relax placements from the last used host to anywhere in the cluster —this allows us to set per-component-type timeouts measurable in minutes. For example HBase masters can have placement relaxed in 20s, while region servers could have a timeout of 10 minutes.

See http://slider.incubator.apache.org/docs/configuration/resource_specification.html for details

Thanks Steve Loughran ! That information is useful.

If I understand correctly, in Slider you are providing best-effort recovery on failure of YARN components like AM crash or RM crash by preserving state in HDFS ?

Hey Navina, this is fantastic! Here are my notes:

1. A half-baked thought: have you considered an alternative strategy where, on a clean shutdown, the SamzaContainer could move its state from the appcache directory to some other directory (e.g. /mnt/u001/stores)? Then, when a SamzaContainer starts back up it, could move the stores from /mnt/u001/stores back into its new appcache directory, and go from there. Not sure if this provides any advantage over just always storing the data in /mnt/u001/stores.
2. I'm not quite sure that I follow your comments on page 2 about samza-standalone. "This leads to re-sharding of the data in the keyvalue stores. This kind of dynamic re-sharding makes providing host-affinity at framework level difficult, esp. because the developer controls the set of machines where the containers will get deployed." It's true that a container failure does cause tasks (and thus stores) to get moved from the failed to container to live containers. Two things I wanted to call out are 1) the stores are not intermingled, and 2) when dead container comes back, it should be able to re-use its local state (if it was not a failed container, but rather a clean shutdown) even if its task was shifted to another machine and made further progress. Can you elaborate a bit on the difficulty you see with standalone?
3. "Number of containers and/or container-partition assignment changes across successive application runs." This isn't a problem, is it? If we had task1, task2, and task3, and we go back and forth between two containers and three containers, shouldn't we shift tasks between them? I think the same thing holds for tasks. Adding a new task shouldn't affect where existing tasks are located, right?
4. Wonder if $STATE_ROOT_DIR should be a config rather than an environment variable. If it's an environment variable, how does it get set?
5. We should try and make this stuff as YARN-agnostic as possible, so we can support it for Mesos as well. I think the container-host mapping, state store location, and container state re-use should be agnostic. Cleaning the state dir and doing the actual AM requests should probably remain YARN-specific. I think that your proposal fits this.
6. Are the FairScheduler configs set in yarn-site.xml, or in some type of scheduler-site.xml file?
7. Rather than including the container-host mapping in the Config object, I think it should be persisted as part of the ContainerModel. This is still served by the JobCoordinator, but is outside of the Config object.
8. Nit: Relocating directory store link is just SAMZA-, with no number.
9. I wonder if the default state store directory should remain the user.dir. This would make new behavior consistent with old. It would also default to disabling state re-use. Using /tmp is a bit dangerous because some systems have limits on the size of a `/tmp` directory. If not, maybe we could use java.io.tmpdir as the default.
10. I think we shouldn't worry too much about per-application retention policies. This whole feature is best effort. We should be a bit pessimistic, and hold on to abandoned data for some configurable amount of time (or until disk starts to run out), and then delete it.

1. A half-baked thought: have you considered an alternative strategy where, on a clean shutdown, the SamzaContainer could move its state from the appcache directory to some other directory (e.g. /mnt/u001/stores)

Yeah. That is an alternative strategy which can be achieved using the NM Aux Services or including the . I don't see any obvious advantage. It can be advantageous if we make store retention optional for the application. We can only move those stores out to /mnt/u001/stores and leave the others in YARN to be garbage collected.

Can you elaborate a bit on the difficulty you see with standalone?

I think my understanding of the standalone design and how the stores are partitioned on the disk was incorrect. I will re-word this. It shouldn't be very different with the standalone use-case.

3. "Number of containers and/or container-partition assignment changes across successive application runs."

Well, this is what I had in mind. In the example:
first attempt of the application -> container 1 (task1, task2) on host1 and container 2 (task3) on host2
second attempt -> container 1 (task1) on host1, container 2 (task2) on host2 and container 3 (task3) on host3
Isn't this case possible? If so, not all partitions will be available on the respective hosts.

Wonder if $STATE_ROOT_DIR should be a config rather than an environment variable. If it's an environment variable, how does it get set?

It is config and not environment variable. I have mentioned this Page 3 of the design under "Relocating store directory" section

6. Are the FairScheduler configs set in yarn-site.xml, or in some type of scheduler-site.xml file?

Well, I think if it is in yarn-site.xml, it will work. There is clearly not enough documentation on the yarn website about this. So, i will have to experiment with it.

7.Rather than including the container-host mapping in the Config object, I think it should be persisted as part of the ContainerModel. This is still served by the JobCoordinator, but is outside of the Config object.

Ok. That makes sense.

8. Nit: Relocating directory store link is just SAMZA-, with no number.

Yeah. Waiting to open a jira for that task and then, fill it in. It does make the document look incomplete though

Using /tmp is a bit dangerous because some systems have limits on the size of a `/tmp` directory. If not, maybe we could use java.io.tmpdir as the default.

Agree

10. I think we shouldn't worry too much about per-application retention policies.

Ok

Navina, we try to re-instantiate components of a distributed app on the same hosts they were on before, supporting best-effort, strict or anywhere placement.

For best-effort & strict we save to HDFS the activity of all nodes on the cluster: whether components were active there, and, if, not, the last time they were used.

when restarting the app, a new YARN app is created, the AM reads in its configs and this history file and uses it to select locations to request the containers it needs.

Failure recovery is a separate process: the AM enumerates the containers, maps from their priority to component type (every component has 2 possible priorities), and so rebuilds its data structures. When the containers heartbeat in (they find the new URLs to the AM via the YARN registry), their state is queried & used to decide whether they will be accepted (if they are live) vs destroyed and re-requested.

BTW, There's a YARN JIRA on an AM being able to register something that will get run in a different container on shutdown, a "cleanup action"

Ah I see. Got it.

In slider you use HDFS to store the status of components and use it for best-effort stickiness.
I guess in Samza we are proposing to persist the status of components (ie. in which host a container is running) in our ConfigStream (usually a Kafka stream). The Job Coordinator then bootstraps from ConfigStream to find the latest config and use that information to request specific container hosts from YARN.

Thanks for elaborating on the placement and failure recovery in slider.

That sounds useful. Let me try to dig out the JIRA.

Hey Steve, I couldn't find the YARN JIRA that you were referring. If you can dig it out, please let me know.

Debugging data stores (SAMZA-)

It is SAMZA-598, right?

If the data store used by an application is persisted on disk beyond the application's lifetime, there has to be a periodic check
to delete unused/orphaned stores. This ensures that the Node Managers (NM) don't run out of disk space.

Another scenario that maybe worth considering is that,

It is SAMZA-598, right?

Yep.

|task 1 locates container_1 in node_1|it reads node_1's task/storename/container_1/OFFSET -> it's actually already out-of-date|

I agree that it will already be out-of-date. That is why we store the "offset" (of the last method in the changelog feed) in the OFFSET file. This way we can sync the changes from "offset" to this existing store, instead of syncing from the offset=0 of the changelog feed.
I can't think of a scenario where this can lead to inconsistency in the data. If you do, please let me know.

This way we can sync the changes from "offset" to this existing store, instead of syncing from the offset=0 of the changelog feed.

Oh, yes, this guarantees the consistency.

YARN-2261 "YARN should have a way to run post-application cleanup"

Thanks, Steve!

Mostly corrected the document based on feedback.
One notable change in the design - how to specify the path for persisting the store (when it is outside of YARN's working directory).
In the earlier design, the location of the store would default to java.io.tmpDir if the STATE_ROOT_DIR was not specified. Now, it would default to YARN's current working directory, thereby disabling local state re-use.
This simplifies the logic of the clean-up script and maintains backward compatibility.