[CONF] Apache Samza > Host affinity in standalone.

[Shanthoosh Venkataraman (Confluence)]

Title: Message Title

Shanthoosh Venkataraman edited a page

Host affinity in standalone. ... Code Block language java - 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 systemstore spec in standalone: 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 language java - {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. 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. 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. ... View page • Add comment • Like

Stop watching space • Manage notifications This message was sent by Atlassian Confluence 5.8.17