SAMZA-2688 [Elasticity] introduce elasticity factor config and key bucket within SSP by lakshmi-manasa-g · Pull Request #1576 · apache/samza

lakshmi-manasa-g · 2022-01-20T23:48:30Z

Feature: Elasticity (SAMZA-2687) for a Samza job allows job to have more tasks than the number of input SystemStreamPartition(SSP). Thus, a job can scale up beyond its input partition count without needing the repartition the input stream.

This is achieved by having elastic tasks which is the same as a task for all practical purposes. But an elastic task consumes only a subset of the messages of an SSP.
With an elasticity factor F (integer), the number of elastic tasks will be F times N with N = original task count.
The F elastic tasks per original task all consume subsets of same SSP as the original task. There will be F subsets (aka key bucket) per SSP and a message falls into an SSP bucket 'i' if its message.key.hash()%F == i.

Changes:

introduce the config for enabling elasticity as job.elasticity.factor. If the job without elasticity has N tasks then with factor = F > 1, there will be F times N (elastic) tasks
Add "key bucket" (an integer ranging 0-F) to SSP which will identify the messages within the SSP
Compute the key bucket the IncomingMessageEnvelope falls into given elasticity factor F.
SamzaObjectMapper changes to serde keyBucket component of SSP.

 Tests: updated unit tests and added new ones.
 
 API Changes: no public API changes

Upgrade Instructions: N/A

Usage Instructions: set the config job.elasticity.factor > 1 to enable elasticity for the job.

…cket within SSP

mynameborat · 2022-01-20T23:56:08Z

samza-api/src/main/java/org/apache/samza/system/SystemStreamPartition.java

    super(system, stream);
    this.partition = partition;
    this.hash = computeHashCode();
+    this.keyBucket = -1;


why not default the keyBucket to just 0 or 1 however you are basing the bucket index? Cut shorts all the additional handling of -1 and so-on and also logically makes sense that in the event of keyBucket < 1 there is one bucket to which all the of the messages belonging to the task is assigned to.

+1, keyBucket can just be 0 by default.

agreed that keeping keyBucket = 0 when there is no elasticity makes this particular SSP class simpler.

my reason for having it to -1 was as follows.
making keyBucket = 0 and removing the check keyBucket == -1 means every time SSP is printed into logs and task model, it will be `SystemStreamPartition [system, stream, partition, 0] with no elasticity enabled which can be confusing.

this check of keyBucket-1 will be restricted to this class and hence i think it is better to have this additional check rather than having to introduce SSP containing 0 in all places.

pl lmk if this makes sense

during the second pr #1580 dev, i realized that there is a bigger problem with having keyBucket=0 in the ssp.tostring when elasticity is not enabled.

GroupBySystemStreamPartition uses ssp.tostring as the taksName and task level metrics (like sends etc) use the task name. keeping keyBucket=0 in all ssp.tostring even when elasticity is not enabled will result in loss of metrics history.

But cant you still have the !=0 check in toString to preserve existing behavior?

yes, that is doable but imo doesnt differ much from the current !=-1 check. pl lmk if you feel strongly about this and i can update.

How exactly is this doable? is your bucket based on 1-index is it? when you have elasticity enabled, one of the bucket will have 0 as the index and hence its representation would be different from others.

yes you are right.currently my indexes are 0-based and hence using -1 as the default when elasticity is off.

if i choose keyBucket=0 for elasticity off, then i will have to start my elasticity enabled indexes from 1,2.. but this does not differ from default when elasticity off = -1 and start indexes from 0,1,2..

rayman7718 · 2022-01-31T16:57:53Z

samza-core/src/main/java/org/apache/samza/config/JobConfig.java

    return getBoolean(CONTAINER_HEARTBEAT_MONITOR_ENABLED, CONTAINER_HEARTBEAT_MONITOR_ENABLED_DEFAULT);
  }
+
+  public boolean getElasticityEnabled() {


not sure why this is needed?

is the question why getElasticityEnabled is needed when we have getElasticityFactor?
callers wanting to know if elasticity is enabled need not and should not be required to know that elasticity.factor = 1 means disabled and >1 means enabled.

rayman7718

couple of minor comments, feel free to fix and ship

mynameborat · 2022-02-04T17:27:43Z

samza-api/src/main/java/org/apache/samza/system/SystemStreamPartition.java

    final int prime = 31;
    int result = super.hashCode();
    result = prime * result + ((partition == null) ? 0 : partition.hashCode());
+    result = prime * result + ((keyBucket == -1) ? 0 : keyBucket);


I just realized this hashing function yields the same hash regardless of elasticity enabled or not for bucket 0 vs -1 respectively.

I feel you could treat 0 as default and modify the toString() to include keyBucket regardless and just regenerate dashboard or make this is API break change in metrics name explicitly as we are anyways going to run into it.

mynameborat

Sync'd up with Rayman! I am fine with sentinel value being either -1 or 0.

SAMZA-2688 [Elasticity] introduce elasticity factor config and key bu…

0e1710b

…cket within SSP

mynameborat reviewed Jan 20, 2022

View reviewed changes

rayman7718 reviewed Jan 31, 2022

View reviewed changes

rayman7718 approved these changes Jan 31, 2022

View reviewed changes

lakshmi-manasa-g mentioned this pull request Feb 1, 2022

SAMZA-2689 [Elasticity] generate JobModel and update RunLoop with (elastic) tasks processing keyBuckets of SystemStreamPartition #1580

Merged

rayman7718 approved these changes Feb 2, 2022

View reviewed changes

mynameborat reviewed Feb 4, 2022

View reviewed changes

rayman7718 approved these changes Feb 4, 2022

View reviewed changes

mynameborat approved these changes Feb 4, 2022

View reviewed changes

mynameborat merged commit 3eaa697 into apache:master Feb 4, 2022

dxichen mentioned this pull request Feb 28, 2022

SAMZA-2723: Fix yarn version and scala version for scalate #1586

Merged

lakshmi-manasa-g mentioned this pull request May 2, 2022

SAMZA-2739: Improve SystemStreamPartition key deserializer to not be rigid about ssp having only 3 fields #1603

Merged

lakshmi-manasa-g mentioned this pull request May 18, 2022

SAMZA-2743: [Elasticity] Add keybucket into SSP serde for checkpoint #1608

Merged

Conversation

lakshmi-manasa-g commented Jan 20, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rayman7718 Jan 31, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rayman7718 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mynameborat left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

rayman7718 Jan 31, 2022 •

edited

Loading