Skip to content

Commit 6067f56

Browse files
mr-arianelabsmr-arianelabs
committed
streaming tuning
1 parent cf9064c commit 6067f56

5 files changed

Lines changed: 110 additions & 79 deletions

File tree

‎hedera-etl-bigquery/src/main/java/com/hedera/etl/StreamingStorageToBigQueryPipeline.java‎

Lines changed: 10 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
*/
2222

2323
import java.util.Map;
24+
import java.util.Objects;
2425
import java.util.Set;
2526

2627
import com.google.api.services.bigquery.model.TableReference;
@@ -178,8 +179,9 @@ void saveAllToBigQuery(Map<String, PCollection<Row>> input, String dataset) {
178179
.withCreateDisposition(createDisposition)
179180
.withWriteDisposition(writeDisposition)
180181
.withoutValidation()
181-
.withTriggeringFrequency(Duration.standardMinutes(1))
182-
.withMethod(BigQueryIO.Write.Method.STORAGE_WRITE_API);
182+
.withExtendedErrorInfo()
183+
// .withTriggeringFrequency(Duration.standardMinutes(1))
184+
.withMethod(BigQueryIO.Write.Method.STREAMING_INSERTS);
183185

184186
if (isDiffTable) {
185187
final var partitioningField =
@@ -213,17 +215,20 @@ void saveAllToBigQuery(Map<String, PCollection<Row>> input, String dataset) {
213215
writer = writer.to(outputTable).useBeamSchema();
214216
}
215217

216-
WriteResult writeResult = rowPCollection.apply("Save %s.%s to BigQuery".formatted(dataset, tableId), writer);
218+
WriteResult writeResult =
219+
rowPCollection
220+
.apply("Filter out nulls", Filter.by(Objects::nonNull))
221+
.apply("Save %s.%s to BigQuery".formatted(dataset, tableId), writer);
217222
writeResult
218-
.getFailedStorageApiInserts()
223+
.getFailedInsertsWithErr()
219224
.apply(
220225
"Log errors",
221226
Filter.by(
222227
err -> {
223228
log.error(
224229
"Failed to insert into table row: {}\nReason:\n{}",
225230
err.getRow(),
226-
err.getErrorMessage());
231+
err.getError());
227232
return true;
228233
}));
229234
}

‎hedera-etl-bigquery/src/main/java/com/hedera/etl/diff/MergeStreamingWithHistory.java‎

Lines changed: 27 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
import java.time.LocalDate;
55
import java.time.ZoneId;
66
import java.time.ZoneOffset;
7+
import java.util.Objects;
78
import java.util.Optional;
89

910
import com.google.common.base.Strings;
@@ -188,28 +189,31 @@ private void saveLatestValues(
188189
PCollection<Row> input, BatchStorageToBigQueryPipelineOptions options) {
189190
final var dataset = options.getTechnicalDataset();
190191

191-
input.apply(
192-
"Save latest to BigQuery",
193-
BigQueryIO.<Row>write()
194-
.to(
195-
element -> {
196-
var tableRef =
197-
HistoryUtil.getTableFor(
198-
dataset,
199-
"%s_latest".formatted(entityName),
200-
LocalDate.ofInstant(
201-
java.time.Instant.ofEpochMilli(element.getTimestamp().getMillis()),
202-
ZoneId.systemDefault()));
203-
204-
return new TableDestination(
205-
tableRef, "Table for %s".formatted(tableRef.getTableId()));
206-
})
207-
.ignoreUnknownValues()
208-
.useBeamSchema()
209-
.withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
210-
.withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE)
211-
.withoutValidation()
212-
.withTriggeringFrequency(Duration.standardHours(1))
213-
.withMethod(BigQueryIO.Write.Method.STORAGE_WRITE_API));
192+
input
193+
.apply("Filter out nulls", Filter.by(Objects::nonNull))
194+
.apply(
195+
"Save latest to BigQuery",
196+
BigQueryIO.<Row>write()
197+
.to(
198+
element -> {
199+
var tableRef =
200+
HistoryUtil.getTableFor(
201+
dataset,
202+
"%s_latest".formatted(entityName),
203+
LocalDate.ofInstant(
204+
java.time.Instant.ofEpochMilli(
205+
element.getTimestamp().getMillis()),
206+
ZoneId.systemDefault()));
207+
208+
return new TableDestination(
209+
tableRef, "Table for %s".formatted(tableRef.getTableId()));
210+
})
211+
.ignoreUnknownValues()
212+
.useBeamSchema()
213+
.withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
214+
.withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
215+
.withoutValidation()
216+
// .withTriggeringFrequency(Duration.standardHours(1))
217+
.withMethod(BigQueryIO.Write.Method.STREAMING_INSERTS));
214218
}
215219
}

‎hedera-etl-bigquery/src/main/java/com/hedera/etl/recordfile/RecordFileTransform.java‎

Lines changed: 14 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -76,20 +76,20 @@ public PCollection<RecordFile> expand(PCollection<MatchResult.Metadata> input) {
7676
.withAllowedTimestampSkew(new Duration(Long.MAX_VALUE)));
7777

7878
return rcdFiles
79-
.apply(
80-
"Add grouping window",
81-
Window.<RecordFileHandler>into(FixedWindows.of(Duration.standardMinutes(1)))
82-
.discardingFiredPanes()
83-
.withAllowedLateness(
84-
Duration.standardHours(1), Window.ClosingBehavior.FIRE_IF_NON_EMPTY)
85-
.withTimestampCombiner(TimestampCombiner.EARLIEST))
86-
.apply("Key by name", WithKeys.of((RecordFileHandler file) -> file.getFilename()))
87-
.setCoder(
88-
KvCoder.of(
89-
NullableCoder.of(StringUtf8Coder.of()), NullableCoder.of(rcdFiles.getCoder())))
90-
.apply("Group by filename", Combine.perKey(RecordFileHandler::merge))
91-
.apply("Reify window to global", Window.into(new GlobalWindows()))
92-
.apply("Drop filenames", Values.create())
79+
// .apply(
80+
// "Add grouping window",
81+
// Window.<RecordFileHandler>into(FixedWindows.of(Duration.standardMinutes(1)))
82+
// .discardingFiredPanes()
83+
// .withAllowedLateness(
84+
// Duration.standardHours(1), Window.ClosingBehavior.FIRE_IF_NON_EMPTY)
85+
// .withTimestampCombiner(TimestampCombiner.EARLIEST))
86+
// .apply("Key by name", WithKeys.of((RecordFileHandler file) -> file.getFilename()))
87+
// .setCoder(
88+
// KvCoder.of(
89+
// NullableCoder.of(StringUtf8Coder.of()), NullableCoder.of(rcdFiles.getCoder())))
90+
// .apply("Group by filename", Combine.perKey(RecordFileHandler::merge))
91+
// .apply("Reify window to global", Window.into(new GlobalWindows()))
92+
// .apply("Drop filenames", Values.create())
9393
.apply(
9494
"Verify signature and parse matching files",
9595
FlatMapElements.into(TypeDescriptor.of(SignedRecordFilesHandler.class))

‎hedera-etl-bigquery/src/main/java/com/hedera/etl/recordfile/Validator.java‎

Lines changed: 39 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,18 @@
11
package com.hedera.etl.recordfile;
22

3+
import java.util.ArrayList;
34
import java.util.Objects;
45
import java.util.Optional;
56
import java.util.concurrent.atomic.AtomicReference;
7+
import java.util.stream.Stream;
68

9+
import com.google.common.collect.Iterables;
710
import lombok.extern.log4j.Log4j2;
811
import org.apache.beam.sdk.coders.InstantCoder;
912
import org.apache.beam.sdk.coders.NullableCoder;
1013
import org.apache.beam.sdk.coders.SerializableCoder;
1114
import org.apache.beam.sdk.coders.StringUtf8Coder;
15+
import org.apache.beam.sdk.state.MapState;
1216
import org.apache.beam.sdk.state.OrderedListState;
1317
import org.apache.beam.sdk.state.StateSpec;
1418
import org.apache.beam.sdk.state.StateSpecs;
@@ -20,6 +24,7 @@
2024
import org.apache.beam.sdk.transforms.DoFn;
2125
import org.apache.beam.sdk.values.KV;
2226
import org.apache.beam.sdk.values.TimestampedValue;
27+
import org.apache.commons.lang3.stream.Streams;
2328
import org.jetbrains.annotations.NotNull;
2429
import org.joda.time.Duration;
2530
import org.joda.time.Instant;
@@ -37,8 +42,8 @@ public ValidatorDoFn(String lastValidHash) {
3742
}
3843

3944
@StateId("previousFiles")
40-
private final StateSpec<OrderedListState<SignedRecordFilesHandler>> previousFiles =
41-
StateSpecs.orderedList(SerializableCoder.of(SignedRecordFilesHandler.class));
45+
private final StateSpec<MapState<Instant, SignedRecordFilesHandler>> previousFiles =
46+
StateSpecs.map(InstantCoder.of(), SerializableCoder.of(SignedRecordFilesHandler.class));
4247

4348
@StateId("lastHash")
4449
private final StateSpec<ValueState<String>> lastHash =
@@ -53,7 +58,7 @@ public ValidatorDoFn(String lastValidHash) {
5358
StateSpecs.value(NullableCoder.of(StringUtf8Coder.of()));
5459

5560
@TimerId("validationTimer")
56-
private final TimerSpec validationTimer = TimerSpecs.timer(TimeDomain.EVENT_TIME);
61+
private final TimerSpec validationTimer = TimerSpecs.timer(TimeDomain.PROCESSING_TIME);
5762

5863
@StateId("timerSet")
5964
private final StateSpec<ValueState<Boolean>> timerSet = StateSpecs.value();
@@ -63,14 +68,14 @@ public ValidatorDoFn(String lastValidHash) {
6368
@ProcessElement
6469
public void processElement(
6570
ProcessContext context,
66-
@StateId("previousFiles") OrderedListState<SignedRecordFilesHandler> previousFiles,
71+
@StateId("previousFiles") MapState<Instant, SignedRecordFilesHandler> previousFiles,
6772
@StateId("timerSet") ValueState<Boolean> timerSet,
6873
@TimerId("validationTimer") Timer timer) {
6974
var value = context.element().getValue();
70-
var timestamp = context.timestamp();
71-
previousFiles.add(TimestampedValue.of(value, timestamp));
75+
var timestamp = value.timestamp();
76+
previousFiles.put(timestamp, value);
7277

73-
log.info(
78+
log.debug(
7479
"Got file {} on nodes {}",
7580
value.getFilename(),
7681
value.getCache().stream()
@@ -83,7 +88,7 @@ public void processElement(
8388
.toList());
8489

8590
if (timerSet.read() != Boolean.TRUE) {
86-
timer.offset(Duration.standardSeconds(10)).setRelative();
91+
timer.offset(Duration.standardSeconds(60)).setRelative();
8792
timerSet.write(true);
8893
}
8994
}
@@ -96,12 +101,14 @@ public Duration getAllowedTimestampSkew() {
96101
@OnTimer("validationTimer")
97102
public void onValidationTimer(
98103
OnTimerContext context,
99-
@StateId("previousFiles") OrderedListState<SignedRecordFilesHandler> previousFiles,
104+
@StateId("previousFiles") MapState<Instant, SignedRecordFilesHandler> previousFiles,
100105
@StateId("lastTs") ValueState<Instant> lastTsState,
101106
@StateId("lastFilename") ValueState<String> lastFilenameState,
102107
@StateId("lastHash") ValueState<String> lastHashState,
103108
@StateId("timerSet") ValueState<Boolean> timerSet) {
104-
var values = previousFiles.read();
109+
log.info("Starting validation/emission round: {}", context.timestamp());
110+
var timestamps = Streams.of(previousFiles.keys().read()).sorted().toList();
111+
log.info("Timestamps in state:\n{}", timestamps);
105112

106113
var firstTs = lastTsState.read();
107114
var lastTs = lastTsState.read();
@@ -115,8 +122,16 @@ public void onValidationTimer(
115122

116123
String badFile = null;
117124

118-
for (var tv : values) {
119-
var filename = tv.getValue().getFilename();
125+
int emittedElements = 0;
126+
127+
for (var ts : timestamps) {
128+
if (emittedElements >= 10000) {
129+
break;
130+
}
131+
132+
var value = previousFiles.get(ts).read();
133+
134+
var filename = value.getFilename();
120135

121136
if (badFile != null && !Objects.equals(filename, badFile)) {
122137
break;
@@ -125,17 +140,21 @@ public void onValidationTimer(
125140

126141
if (Objects.equals(filename, lastFilename)) {
127142
log.info("Ignoring duplicate of file {}", lastFilename);
143+
previousFiles.remove(ts);
128144
continue;
129145
}
130146

131-
var ts = tv.getTimestamp();
132-
if (firstTs != null && ts.isBefore(firstTs)) {
147+
var isLateFile = firstTs != null && ts.isBefore(firstTs);
148+
if (isLateFile) {
133149
log.info("Ignoring late file {}", lastFilename);
150+
previousFiles.remove(ts);
134151
continue;
135152
}
136153

154+
log.info("Validating file {}", filename);
155+
137156
var validRecordFile =
138-
tv.getValue().getCache().stream()
157+
value.getCache().stream()
139158
.filter(
140159
recordFile -> {
141160
var previousHash = recordFile.getPreviousHash();
@@ -167,16 +186,19 @@ public void onValidationTimer(
167186
lastHash.set(recordFile.getHash());
168187
log.info("Emitting file {}", lastFilename);
169188
context.outputWithTimestamp(recordFile, ts);
189+
emittedElements++;
190+
previousFiles.remove(ts);
170191
}
171192
}
172193

173194
if (firstTs != null && lastTs != null) {
174-
log.info("Clearing record files between {} - {}", firstTs, lastTs);
175-
previousFiles.clearRange(firstTs, lastTs.plus(Duration.millis(1)));
195+
log.info("Emitted record files between {} - {}", firstTs, lastTs);
176196
}
197+
lastTsState.write(lastTs);
177198
lastHashState.write(lastHash.get());
178199
lastFilenameState.write(lastFilename);
179200
timerSet.write(false);
201+
log.info("Finishing validation/emission round: {}. Last filename: {}", context.timestamp(), lastFilename);
180202
}
181203
}
182204
}

‎hedera-etl-bigquery/src/test/java/com/hedera/etl/recordfile/ValidatorTest.java‎

Lines changed: 20 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -27,10 +27,10 @@ public void testPreviousHashState() {
2727
.apply(
2828
Create.timestamped(
2929
List.of(
30-
new SignedRecordFileStub("file0", "hash0", "---"),
31-
new SignedRecordFileStub("file1", "hash1", "hash0"),
32-
new SignedRecordFileStub("file2", "hash2", "hash1"),
33-
new SignedRecordFileStub("file3", "hash3", "hash2")),
30+
new SignedRecordFileStub("1970-01-01T00_00_00.000000000Z.rcd", "hash0", "---"),
31+
new SignedRecordFileStub("1970-01-01T00_01_00.000000000Z.rcd", "hash1", "hash0"),
32+
new SignedRecordFileStub("1970-01-01T00_02_00.000000000Z.rcd", "hash2", "hash1"),
33+
new SignedRecordFileStub("1970-01-01T00_03_00.000000000Z.rcd", "hash3", "hash2")),
3434
List.of(0L, 1L, 2L, 3L).stream()
3535
.map(x -> Duration.ofMinutes(x).toMillis())
3636
.toList()))
@@ -42,7 +42,7 @@ public void testPreviousHashState() {
4242
.apply("Global", Window.into(new GlobalWindows()))
4343
.apply(MapElements.into(TypeDescriptors.strings()).via(f -> f.getName()));
4444

45-
PAssert.that(result).containsInAnyOrder("file0", "file1", "file2", "file3");
45+
PAssert.that(result).containsInAnyOrder("1970-01-01T00_00_00.000000000Z.rcd", "1970-01-01T00_01_00.000000000Z.rcd", "1970-01-01T00_02_00.000000000Z.rcd", "1970-01-01T00_03_00.000000000Z.rcd");
4646

4747
pipeline.run();
4848
}
@@ -54,11 +54,11 @@ public void testPreviousHashStateWithDuplicates() {
5454
.apply(
5555
Create.timestamped(
5656
List.of(
57-
new SignedRecordFileStub("file0", "hash0", "---"),
58-
new SignedRecordFileStub("file1", "hash1", "hash0"),
59-
new SignedRecordFileStub("file2", "hash2", "hash1"),
60-
new SignedRecordFileStub("file2", "hash2", "hash1"),
61-
new SignedRecordFileStub("file3", "hash3", "hash2")),
57+
new SignedRecordFileStub("1970-01-01T00_00_00.000000000Z.rcd", "hash0", "---"),
58+
new SignedRecordFileStub("1970-01-01T00_01_00.000000000Z.rcd", "hash1", "hash0"),
59+
new SignedRecordFileStub("1970-01-01T00_02_00.000000000Z.rcd", "hash2", "hash1"),
60+
new SignedRecordFileStub("1970-01-01T00_02_00.000000000Z.rcd", "hash2", "hash1"),
61+
new SignedRecordFileStub("1970-01-01T00_03_00.000000000Z.rcd", "hash3", "hash2")),
6262
List.of(0L, 1L, 2L, 2L, 3L).stream()
6363
.map(x -> Duration.ofMinutes(x).toMillis())
6464
.toList()))
@@ -70,7 +70,7 @@ public void testPreviousHashStateWithDuplicates() {
7070
.apply("Global", Window.into(new GlobalWindows()))
7171
.apply(MapElements.into(TypeDescriptors.strings()).via(f -> f.getName()));
7272

73-
PAssert.that(result).containsInAnyOrder("file0", "file1", "file2", "file3");
73+
PAssert.that(result).containsInAnyOrder("1970-01-01T00_00_00.000000000Z.rcd", "1970-01-01T00_01_00.000000000Z.rcd", "1970-01-01T00_02_00.000000000Z.rcd", "1970-01-01T00_03_00.000000000Z.rcd");
7474

7575
pipeline.run();
7676
}
@@ -82,14 +82,14 @@ public void testPreviousHashStateWithDuplicatesWhereOneIsBad() {
8282
.apply(
8383
Create.timestamped(
8484
List.of(
85-
new SignedRecordFileStub("file0", "hash0", "---"),
86-
new SignedRecordFileStub("file1", "hash1", "hash0"),
87-
new SignedRecordFileStub("file2", "bad-hash2", "bad-hash1"),
88-
new SignedRecordFileStub("file2", "bad-hash2", "bad-hash1"),
89-
new SignedRecordFileStub("file2", "bad-hash2", "bad-hash1"),
90-
new SignedRecordFileStub("file2", "bad-hash2", "bad-hash1"),
91-
new SignedRecordFileStub("file2", "hash2", "hash1"),
92-
new SignedRecordFileStub("file3", "hash3", "hash2")),
85+
new SignedRecordFileStub("1970-01-01T00_00_00.000000000Z.rcd", "hash0", "---"),
86+
new SignedRecordFileStub("1970-01-01T00_01_00.000000000Z.rcd", "hash1", "hash0"),
87+
new SignedRecordFileStub("1970-01-01T00_02_00.000000000Z.rcd", "bad-hash2", "bad-hash1"),
88+
new SignedRecordFileStub("1970-01-01T00_02_00.000000000Z.rcd", "bad-hash2", "bad-hash1"),
89+
new SignedRecordFileStub("1970-01-01T00_02_00.000000000Z.rcd", "bad-hash2", "bad-hash1"),
90+
new SignedRecordFileStub("1970-01-01T00_02_00.000000000Z.rcd", "bad-hash2", "bad-hash1"),
91+
new SignedRecordFileStub("1970-01-01T00_02_00.000000000Z.rcd", "hash2", "hash1"),
92+
new SignedRecordFileStub("1970-01-01T00_03_00.000000000Z.rcd", "hash3", "hash2")),
9393
List.of(0L, 1L, 2L, 2L, 2L, 2L, 2L, 3L).stream()
9494
.map(x -> Duration.ofMinutes(x).toMillis())
9595
.toList()))
@@ -101,7 +101,7 @@ public void testPreviousHashStateWithDuplicatesWhereOneIsBad() {
101101
.apply("Global", Window.into(new GlobalWindows()))
102102
.apply(MapElements.into(TypeDescriptors.strings()).via(f -> f.getName()));
103103

104-
PAssert.that(result).containsInAnyOrder("file0", "file1", "file2", "file3");
104+
PAssert.that(result).containsInAnyOrder("1970-01-01T00_00_00.000000000Z.rcd", "1970-01-01T00_01_00.000000000Z.rcd", "1970-01-01T00_02_00.000000000Z.rcd", "1970-01-01T00_03_00.000000000Z.rcd");
105105

106106
pipeline.run();
107107
}

0 commit comments

Comments
 (0)