| ... If an optimal assignment for each task to a particular processor is generated in the JobModel as part of the leader in a stateful processors group, each follower will just pick up their assignments from job model after the rebalance phase and start processing(similar to non-stateful jobs). The goal is to guarantee that the optimal assignment happens which minimizes the task movement between the processors. Local state of the tasks will be persisted in a directory(local.store.dir) provided through configuration by each processor. ![Standalone host affinity]() Image Modified ZK Data Model to support host affinity:
![]() Code Block |
|
|
- zkBaseRootPath/$appName-$appId-$JobName-$JobId-$stageId/
- processors/
- processor.000001/
- physicalHost1
- processor.000002/
- physicalHost2
...
- processor.00000N/
- physicalHostN
- jobModelVersion/
- {$Generation.$Version}
- JobModel
- JobConfiguration
- ContainerModel-1
- processorId: 0
- hostName: physicalHostName-1
- TaskModel-1
- taskName: TaskId-1
- SystemStreamPartitions: [InputStream1:P1, InputStream1:P2]
- changeLogPartitionId: 0
- TaskModel-2
- taskName: TaskId-2
- systemStreamPartitions: [InputStream1:P3, InputStream1:P4]
- ChangeLogPartitionId: 1
- ContainerModel-2
- processorId: 1
- hostName: physicalHostName-2
- TaskModel-3
- taskName: TaskId-3
- SystemStreamPartitions: [InputStream1:P5, InputStream1:P6]
- ChangeLogPartitionId: 2
- TaskModel-4
- taskName: TaskId-4
- SystemStreamPartitions: [InputStream1:P7, InputStream1:P8]
- ChangeLogPartitionId: 3 |
Local file system:
In standalone landscape, the file system location to persist the local state should be provided by the users through stream processor configuration(by defining local.store.dir configuration). It’s expected that the stream processor java process will be configured by user to run with sufficient read/write permissions to access the local state directories created by any processor in the group. Users should manage the garbage collection of unused and stale local state’s created by the stream processors. ...
When low level StreamTask/AsyncStreamTask API is used:
![]() Code Block |
|
|
- {local.store.dir}
- {jobId-jobName-1}
- {taskName-1}
- store-1
- OFFSET
- 1.sst
- {jobId-jobName-2}
- {taskName-2}
- store-2
- OFFSET
- 1.sst |
When high level fluent API is used:
![]() Code Block |
- {local.store.dir}
- {appId-appName-1}
- {taskName-1}
- {operatorId-1}
- store1
- OFFSET
- 1.sst
- {taskName-2}
- {operatorId-2}
- store2
- OFFSET
- 1.sst |
State store restoration:
In order to make local store data highly available, all writes to the local store are replicated to a durable changelog(kafka topic) which serves the purpose of replication log. Each local store has an associated OFFSET file which represents the local state store lag behind the changelog(source of truth). When a samza task commits or samza container shuts down, pending local store changes are flushed to change log stream and latest changelog offset is persisted to OFFSET file on disk.
Upon processor restart, if the local state associated with the task is not available on the disk, it will be restored following the existing restoration logic followed in yarn deployment model. The following represents the state store restoration sequence which occurs upon Stream processor restart:
-
Opens the persisted store on disk.
-
Reads the OFFSET file.
-
Restores the state store from the OFFSET value.
When assigning tasks to a stream processor in a run, the stream processor to which the task was assigned in the previous run will be preferred. If the stream processor to which task was assigned in previous run is unavailable in the current run, the stream processors running on physical host of previous run will be given higher priority and favored. If both of the above two conditions are not met, then the task will be assigned to any stream processor available in the processor group.
2. Remove coordinator stream bindings from JobModel:
JobModel is a data access object used to represent a samza job in both yarn and standalone deployment models. With existing implementation, JobModel requires LocalityManager(which is tied to coordinator stream) to read and populate processor locality assignments. However, since zookeeper is used as JobModel persistence layer and coordinator stream doesn’t exist in standalone landscape, it’s essential to remove this LocalityManager binding from JobModel and make JobModel immutable. It will be ideal to store task to preferred host assignment as a part of job model due to the following reasons:
-
Task to locality assignment information logically belongs to the JobModel itself and it makes things simpler by persisting them together.
-
If task to preferred host assignment and JobModel are stored separately in zookeeper in standalone, we’ll run into consistency problems between these two data sinks when performing JobModel upgrades. We’ll also lose the capability to do atomic updates of entire JobModel in zookeeper.
Any existing implementations(ClusterBasedJobCoordinator, ContainerProcessManager) which depends upon this binding for functional correctness in samza-yarn, should directly read container locality from the coordinator stream instead of getting it indirectly via JobModel.
3. Clean Cleaning up ContainerModel:
ContainerModel is a data access object used in samza for holding the task to system stream partition assignments which is generated by TaskNameGrouper implementations. ContainerModel currently has two fields(processorId and containerID) used to uniquely identify a processor in a processors group. Standalone deployment model uses processorId and Yarn deployment model uses containerId field to store the unique processorId. To achieve uniformity between the two deployment models, the proposal is to remove duplicate containerId. This will not require any operational migration.
Public Interfaces Existing interface:
![]() Code Block |
|
|
public interface TaskNameGrouper {
Set<ContainerModel> group(Set<TaskModel> tasks);
default Set<ContainerModel> group(Set<TaskModel> tasks, List<String> containersIds) {
return group(tasks);
}
}
public interface BalancingTaskNameGrouper extends TaskNameGrouper {
Set<ContainerModel> balance(Set<TaskModel> tasks, LocalityManager localityManager);
}
public class ContainerModel {
@Deprecated
private final int containerId;
private final String processorId;
private final Map<TaskName, TaskModel> tasks; |
...
![]() Code Block |
public interface TaskNameGrouper {
@deprecated
Set<ContainerModel> group(Set<TaskModel> tasks);
@deprecated
default Set<ContainerModel> group(Set<TaskModel> tasks, List<String> containersIds) {
return group(tasks);
}
Set<ContainerModel> group(Set<TaskModel> currentGenerationTaskModels, List<String> currentGenerationContainerIds, Set<ContainerModel> previousGenerationContainerModels);
}
public interface BalancingTaskNameGrouper extends TaskNameGrouper {
@deprecated
Set<ContainerModel> balance(Set<TaskModel> tasks, LocalityManager localityManager);
}
public class ContainerModel {
// ContainerId field is deleted.
private final String processorId;
private final Map<TaskName, TaskModel> tasks;
// New field added denoting the physical hostname.
private final String hostName; |
Here are few reasons supporting the modification of TaskNameGrouper interface and removing LocalityManager from interface methods:
-
Any TaskNameGrouper implementation should be usable in both yarn and standalone deployment models. However, TaskNameGrouper interface definition has an explicit and tight binding with CoordinatorStream through LocalityManager and existing TaskNameGrouper implementations employs LocalityManager to read/write locality mapping from and to Coordinator stream. Coordinator stream doesn’t exist in standalone landscape and this prohibits usage of some TaskNameGrouper implementations in standalone.
-
Multiple group methods in TaskNameGrouper interface and additional balance method in BalancingTaskNameGrouper are logically synonymous to each other and exists to generate ContainerModels based upon the input task models and past locality assignments. It’s sensible to combine them into one interface method with adequate parameters and simplify things.
-
Any future TaskNameGrouper implementation could hold references to LocalityManager(a live object) and create object hierarchies based upon that reference. This will clutter the ownership of LocalityManager and could potentially create an unintentional resource leak.
-
Number of processors is a static configuration in yarn deployment model and a job restart is required to change the number of processors. However, an addition/deletion of a processor to a processors group in standalone is quite common and an expected behavior. Existing generators discard the task to physical host assignment when generating the JobModel. However, for standalone it’s essential to consider this detail(task to physical host assignment) between successive job model generations to accomplish optimal task to processor assignment. For instance, let’s assume stream processors P1, P2 runs on host H1 and processor P3 runs on host H3. If P1 dies, it is optimal to assign some of the tasks processed by P1 to P2. If previous task to physical host assignment is not taken into account when generating JobModel, this cannot be achieved.
-
Logically, a TaskNameGrouper implementation would just require the previous generation container models(to get previous task to preferred host mapping, previous task to systemstreampartition mapping) which can be passed in through the interface method to generate new mapping. Any modifications to existing assignments should be done outside of TaskNameGrouper implementation. This will make any implementation as a pure function simply operating on the passed in data.
After this change, we will have one method in TaskNameGrouper interface clearly defining the contract and all other methods in TaskNameGrouper will be deprecated(eventually removed). Host aware task to stream processors assignment in standalone will be housed in a TaskNameGrouper implementation which will be used to support this feature. Implementation and Test Plan ... |
