Skip to content

[Transform] provide exponential_avg* stats for batch transforms #52041

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 8 commits into from
Feb 12, 2020
Merged

[Transform] provide exponential_avg* stats for batch transforms #52041

Show file tree
Hide file tree
Changes from 7 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
ffe2f02
apply spotless code formating
Feb 7, 2020
ea60c09
omit continuous stats for batch transforms
Feb 7, 2020
e57690a
Revert "omit continuous stats for batch transforms"
Feb 7, 2020
fd7cadc
remove extra constructor and always calculate exponential averages
Feb 7, 2020
05322a8
fix BWC test
Feb 10, 2020
f74d2c8
fix more tests
Feb 10, 2020
f9029b0
apply spotless
Feb 10, 2020
8ba947d
fix placement of docstring
Feb 10, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
120 changes: 87 additions & 33 deletions ...rc/main/java/org/elasticsearch/xpack/core/transform/transforms/TransformIndexerStats.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@

import org.elasticsearch.Version;
import org.elasticsearch.common.ParseField;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.xcontent.ConstructingObjectParser;
Expand Down Expand Up @@ -36,23 +37,34 @@ public class TransformIndexerStats extends IndexerJobStats {
public static ParseField SEARCH_TOTAL = new ParseField("search_total");
public static ParseField SEARCH_FAILURES = new ParseField("search_failures");
public static ParseField INDEX_FAILURES = new ParseField("index_failures");
public static ParseField EXPONENTIAL_AVG_CHECKPOINT_DURATION_MS =
new ParseField("exponential_avg_checkpoint_duration_ms");
public static ParseField EXPONENTIAL_AVG_DOCUMENTS_INDEXED =
new ParseField("exponential_avg_documents_indexed");
public static ParseField EXPONENTIAL_AVG_DOCUMENTS_PROCESSED =
new ParseField("exponential_avg_documents_processed");
public static ParseField EXPONENTIAL_AVG_CHECKPOINT_DURATION_MS = new ParseField("exponential_avg_checkpoint_duration_ms");
public static ParseField EXPONENTIAL_AVG_DOCUMENTS_INDEXED = new ParseField("exponential_avg_documents_indexed");
public static ParseField EXPONENTIAL_AVG_DOCUMENTS_PROCESSED = new ParseField("exponential_avg_documents_processed");

// This changes how much "weight" past calculations have.
// The shorter the window, the less "smoothing" will occur.
private static final int EXP_AVG_WINDOW = 10;
private static final double ALPHA = 2.0/(EXP_AVG_WINDOW + 1);
private static final double ALPHA = 2.0 / (EXP_AVG_WINDOW + 1);

private static final ConstructingObjectParser<TransformIndexerStats, Void> LENIENT_PARSER = new ConstructingObjectParser<>(
NAME, true,
args -> new TransformIndexerStats(
(long) args[0], (long) args[1], (long) args[2], (long) args[3], (long) args[4], (long) args[5], (long) args[6],
(long) args[7], (long) args[8], (long) args[9], (Double) args[10], (Double) args[11], (Double) args[12]));
NAME,
true,
args -> new TransformIndexerStats(
(long) args[0],
(long) args[1],
(long) args[2],
(long) args[3],
(long) args[4],
(long) args[5],
(long) args[6],
(long) args[7],
(long) args[8],
(long) args[9],
(Double) args[10],
(Double) args[11],
(Double) args[12]
)
);

static {
LENIENT_PARSER.declareLong(constructorArg(), NUM_PAGES);
Expand All @@ -73,37 +85,62 @@ public class TransformIndexerStats extends IndexerJobStats {
private double expAvgCheckpointDurationMs;
private double expAvgDocumentsIndexed;
private double expAvgDocumentsProcessed;

/**
* Create with all stats set to zero
*/
public TransformIndexerStats() {
super();
}

public TransformIndexerStats(long numPages, long numInputDocuments, long numOutputDocuments,
long numInvocations, long indexTime, long searchTime, long indexTotal, long searchTotal,
long indexFailures, long searchFailures, Double expAvgCheckpointDurationMs,
Double expAvgDocumentsIndexed, Double expAvgDocumentsProcessed ) {
super(numPages, numInputDocuments, numOutputDocuments, numInvocations, indexTime, searchTime, indexTotal, searchTotal,
indexFailures, searchFailures);
public TransformIndexerStats(
long numPages,
long numInputDocuments,
long numOutputDocuments,
long numInvocations,
long indexTime,
long searchTime,
long indexTotal,
long searchTotal,
long indexFailures,
long searchFailures,
Double expAvgCheckpointDurationMs,
Double expAvgDocumentsIndexed,
Double expAvgDocumentsProcessed
) {
super(
numPages,
numInputDocuments,
numOutputDocuments,
numInvocations,
indexTime,
searchTime,
indexTotal,
searchTotal,
indexFailures,
searchFailures
);
this.expAvgCheckpointDurationMs = expAvgCheckpointDurationMs == null ? 0.0 : expAvgCheckpointDurationMs;
this.expAvgDocumentsIndexed = expAvgDocumentsIndexed == null ? 0.0 : expAvgDocumentsIndexed;
this.expAvgDocumentsProcessed = expAvgDocumentsProcessed == null ? 0.0 : expAvgDocumentsProcessed;
}

public TransformIndexerStats(long numPages, long numInputDocuments, long numOutputDocuments,
long numInvocations, long indexTime, long searchTime, long indexTotal, long searchTotal,
long indexFailures, long searchFailures) {
this(numPages, numInputDocuments, numOutputDocuments, numInvocations, indexTime, searchTime, indexTotal, searchTotal,
indexFailures, searchFailures, 0.0, 0.0, 0.0);
}

Copy link
Author

@hendrikmuhs hendrikmuhs Feb 10, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

^ removed this constructor

public TransformIndexerStats(TransformIndexerStats other) {
this(other.numPages, other.numInputDocuments, other.numOuputDocuments, other.numInvocations,
other.indexTime, other.searchTime, other.indexTotal, other.searchTotal, other.indexFailures, other.searchFailures);
this.expAvgCheckpointDurationMs = other.expAvgCheckpointDurationMs;
this.expAvgDocumentsIndexed = other.expAvgDocumentsIndexed;
this.expAvgDocumentsProcessed = other.expAvgDocumentsProcessed;
this(
other.numPages,
other.numInputDocuments,
other.numOuputDocuments,
other.numInvocations,
other.indexTime,
other.searchTime,
other.indexTotal,
other.searchTotal,
other.indexFailures,
other.searchFailures,
other.expAvgCheckpointDurationMs,
other.expAvgDocumentsIndexed,
other.expAvgDocumentsProcessed
);
}

public TransformIndexerStats(StreamInput in) throws IOException {
Expand Down Expand Up @@ -180,7 +217,7 @@ public void incrementCheckpointExponentialAverages(long checkpointDurationMs, lo
}

private double calculateExpAvg(double previousExpValue, double alpha, long observedValue) {
return alpha * observedValue + (1-alpha) * previousExpValue;
return alpha * observedValue + (1 - alpha) * previousExpValue;
}

@Override
Expand Down Expand Up @@ -212,9 +249,26 @@ public boolean equals(Object other) {

@Override
public int hashCode() {
return Objects.hash(numPages, numInputDocuments, numOuputDocuments, numInvocations,
indexTime, searchTime, indexFailures, searchFailures, indexTotal, searchTotal,
expAvgCheckpointDurationMs, expAvgDocumentsIndexed, expAvgDocumentsProcessed);
return Objects.hash(
numPages,
numInputDocuments,
numOuputDocuments,
numInvocations,
indexTime,
searchTime,
indexFailures,
searchFailures,
indexTotal,
searchTotal,
expAvgCheckpointDurationMs,
expAvgDocumentsIndexed,
expAvgDocumentsProcessed
);
}

@Override
public String toString() {
return Strings.toString(this);
}

public static TransformIndexerStats fromXContent(XContentParser parser) {
Expand Down
16 changes: 12 additions & 4 deletions ...st/java/org/elasticsearch/xpack/core/transform/transforms/TransformIndexerStatsTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,13 +31,21 @@ protected TransformIndexerStats doParseInstance(XContentParser parser) {
}

public static TransformIndexerStats randomStats() {
return new TransformIndexerStats(randomLongBetween(10L, 10000L),
randomLongBetween(0L, 10000L), randomLongBetween(0L, 10000L), randomLongBetween(0L, 10000L), randomLongBetween(0L, 10000L),
randomLongBetween(0L, 10000L), randomLongBetween(0L, 10000L), randomLongBetween(0L, 10000L), randomLongBetween(0L, 10000L),
return new TransformIndexerStats(
randomLongBetween(10L, 10000L),
randomLongBetween(0L, 10000L),
randomLongBetween(0L, 10000L),
randomLongBetween(0L, 10000L),
randomLongBetween(0L, 10000L),
randomLongBetween(0L, 10000L),
randomLongBetween(0L, 10000L),
randomLongBetween(0L, 10000L),
randomLongBetween(0L, 10000L),
randomLongBetween(0L, 10000L),
randomBoolean() ? randomDouble() : null,
randomBoolean() ? randomDouble() : null,
randomBoolean() ? randomDouble() : null);
randomBoolean() ? randomDouble() : null
);
}

public void testExpAvgIncrement() {
Expand Down
2 changes: 1 addition & 1 deletion .../src/test/java/org/elasticsearch/xpack/core/transform/transforms/TransformStatsTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,7 @@ public void testBwcWith73() throws IOException {
STARTED,
randomBoolean() ? null : randomAlphaOfLength(100),
randomBoolean() ? null : NodeAttributeTests.randomNodeAttributes(),
new TransformIndexerStats(1, 2, 3, 4, 5, 6, 7, 8, 9, 10),
new TransformIndexerStats(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 0.0, 0.0, 0.0),
Copy link
Author

@hendrikmuhs hendrikmuhs Feb 10, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

do not get confused, testBwcWith73: this test BWC with 7.3 where we had no exponential averages, so its ok to construct the stats with 0.0

new TransformCheckpointingInfo(
new TransformCheckpointStats(0, null, null, 10, 100),
new TransformCheckpointStats(0, null, null, 100, 1000),
Expand Down
Loading