You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2022/03/15 22:29:55 UTC
[GitHub] [beam] egalpin opened a new pull request #17097: [BEAM-14064] ElasticsearchIO remove bundle based
egalpin opened a new pull request #17097:
URL: https://github.com/apache/beam/pull/17097
Removes bundle-based batching for ElasticsearchIO in favour of solely supporting state-based batching (via GroupIntoBatches). This will ensure that batching has a centralized transform for re-use, as well as ensuring that windows stay intact throughout ElasticsearchIO#write.
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] je-ik commented on pull request #17097: [BEAM-14064] ElasticsearchIO remove bundle based
Posted by GitBox <gi...@apache.org>.
je-ik commented on pull request #17097:
URL: https://github.com/apache/beam/pull/17097#issuecomment-1069166593
> b) For anyone using stateful batching before, no performance change would be present. In terms of _Elasticsearch_ performance, state-based batching is highly preferred in my experience. I have been able to improve throughput 100x by using state-based over bundle-based batching. In terms of Beam performance on a runner, I don't have concrete numbers describing the performance delta. I do know that a production workload I manage processes millions of documents (GBs of data) per hour on a single vCPU using stateful batching. In my experience, the process of indexing data into ES has always been heavily IO bound.
Makes sense. Thanks for clarification.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] egalpin commented on pull request #17097: [BEAM-14064] ElasticsearchIO remove bundle based
Posted by GitBox <gi...@apache.org>.
egalpin commented on pull request #17097:
URL: https://github.com/apache/beam/pull/17097#issuecomment-1069160581
@je-ik thanks for the review!
> Generally LGTM. I have a few comments. Two concerns:
a) the ElasticsearchIO now does batching by default, are we sure, that users that did not do the batching previously will be able to successfully upgrade the running pipeline? I think that probably yes, but did we test that?
b) do we know how this change affects performance?
a) The Elasticsearch#write transform has always done batching, but initial implementation was bundle-based only. I agree though that this change set should be tested against a running pipeline where the pipeline is updated; I can take that on and report back.
b) For anyone using stateful batching before, no performance change would be present. In terms of _Elasticsearch_ performance, state-based batching is highly preferred in my experience. I have been able to improve throughput 100x by using state-based over bundle-based batching. In terms of Beam performance on a runner, I don't have concrete numbers describing the performance delta. I do know that a production workload I manage processes millions of documents (GBs of data) per hour on a single vCPU using stateful batching. In my experience, the process of indexing data into ES has always been heavily IO bound.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] egalpin closed pull request #17097: [BEAM-14064] ElasticsearchIO remove bundle based
Posted by GitBox <gi...@apache.org>.
egalpin closed pull request #17097:
URL: https://github.com/apache/beam/pull/17097
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] egalpin commented on a change in pull request #17097: [BEAM-14064] ElasticsearchIO remove bundle based
Posted by GitBox <gi...@apache.org>.
egalpin commented on a change in pull request #17097:
URL: https://github.com/apache/beam/pull/17097#discussion_r828024877
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupIntoBatches.java
##########
@@ -221,6 +221,31 @@ private GroupIntoBatches(BatchingParams<InputT> params) {
duration));
}
+ public GroupIntoBatches<K, InputT> withByteSize(Long batchSizeBytes) {
Review comment:
Just misuse on my part, I thought I'd copied what was done in the other methods. Changed in a recent commit 👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] echauchot commented on pull request #17097: [BEAM-14064] ElasticsearchIO remove bundle based
Posted by GitBox <gi...@apache.org>.
echauchot commented on pull request #17097:
URL: https://github.com/apache/beam/pull/17097#issuecomment-1070534414
> > b) For anyone using stateful batching before, no performance change would be present. In terms of _Elasticsearch_ performance, state-based batching is highly preferred in my experience. I have been able to improve throughput 100x by using state-based over bundle-based batching. In terms of Beam performance on a runner, I don't have concrete numbers describing the performance delta. I do know that a production workload I manage processes millions of documents (GBs of data) per hour on a single vCPU using stateful batching. In my experience, the process of indexing data into ES has always been heavily IO bound.
>
> Makes sense. Thanks for clarification.
Agree but state support in streaming mode does not work with spark RDD/DataStream runner. So in some cases, still needs to do in memory batching. And state is still unimplemented for Spark Structured Streaming runner.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] je-ik commented on a change in pull request #17097: [BEAM-14064] ElasticsearchIO remove bundle based
Posted by GitBox <gi...@apache.org>.
je-ik commented on a change in pull request #17097:
URL: https://github.com/apache/beam/pull/17097#discussion_r827786188
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupIntoBatches.java
##########
@@ -221,6 +221,31 @@ private GroupIntoBatches(BatchingParams<InputT> params) {
duration));
}
+ public GroupIntoBatches<K, InputT> withByteSize(Long batchSizeBytes) {
Review comment:
This probably deserves a javadoc comment. Why do we use big `Long` instead of `long`?
##########
File path: sdks/java/io/elasticsearch/src/main/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIO.java
##########
@@ -2335,126 +2270,40 @@ public void setup() throws IOException {
}
}
- @StartBundle
- public void startBundle(StartBundleContext context) {
- batch = ArrayListMultimap.create();
- currentBatchSizeBytes = 0;
- }
-
- /**
- * Adapter interface which provides a common parent for {@link ProcessContext} and {@link
- * FinishBundleContext} so that we are able to use a single common invocation to output from.
- */
- interface ContextAdapter {
- void output(
- TupleTag<Document> tag, Document document, Instant timestamp, BoundedWindow window);
- }
-
- private static final class ProcessContextAdapter<T> implements ContextAdapter {
- private final DoFn<T, Document>.ProcessContext context;
-
- private ProcessContextAdapter(DoFn<T, Document>.ProcessContext context) {
- this.context = context;
- }
-
- @Override
- public void output(
- TupleTag<Document> tag, Document document, Instant ignored1, BoundedWindow ignored2) {
- // Note: window and timestamp are intentionally unused, but required as params to fit the
- // interface
- context.output(tag, document);
- }
- }
-
- private static final class FinishBundleContextAdapter<T> implements ContextAdapter {
- private final DoFn<T, Document>.FinishBundleContext context;
-
- private FinishBundleContextAdapter(DoFn<T, Document>.FinishBundleContext context) {
- this.context = context;
- }
-
- @Override
- public void output(
- TupleTag<Document> tag, Document document, Instant timestamp, BoundedWindow window) {
- context.output(tag, document, timestamp, window);
+ @ProcessElement
+ public void processElement(ProcessContext c) throws IOException, InterruptedException {
Review comment:
Can we avoid `ProcessContext` and use `@Element` and `OutputCollector` for new DoFns? Might be worth changing for the old ones as well, but at least for the new code, we might want to use the new style.
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupIntoBatches.java
##########
@@ -221,6 +221,31 @@ private GroupIntoBatches(BatchingParams<InputT> params) {
duration));
}
+ public GroupIntoBatches<K, InputT> withByteSize(Long batchSizeBytes) {
+ checkArgument(
+ batchSizeBytes != null && batchSizeBytes < Long.MAX_VALUE && batchSizeBytes > 0,
+ "batchSizeBytes should be a non-negative value less than " + Long.MAX_VALUE);
+ return new GroupIntoBatches<>(
+ BatchingParams.create(
+ params.getBatchSize(),
+ batchSizeBytes,
+ params.getElementByteSize(),
+ params.getMaxBufferingDuration()));
+ }
+
+ public GroupIntoBatches<K, InputT> withByteSize(
Review comment:
Please add javadoc, same for `Long` vs `long` as above.
##########
File path: sdks/java/io/elasticsearch-tests/elasticsearch-tests-common/src/test/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIOTestCommon.java
##########
@@ -336,11 +333,11 @@ void testWriteWithErrors() throws Exception {
ElasticsearchIO.write()
.withConnectionConfiguration(connectionConfiguration)
.withMaxBatchSize(BATCH_SIZE);
- List<String> input =
- ElasticsearchIOTestUtils.createDocuments(
- numDocs, ElasticsearchIOTestUtils.InjectionMode.INJECT_SOME_INVALID_DOCS);
+ // List<String> input =
+ // ElasticsearchIOTestUtils.createDocuments(
+ // numDocs, ElasticsearchIOTestUtils.InjectionMode.INJECT_SOME_INVALID_DOCS);
Review comment:
Is this some left-over?
##########
File path: sdks/java/io/elasticsearch-tests/elasticsearch-tests-common/src/test/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIOTestCommon.java
##########
@@ -362,11 +359,17 @@ public boolean matches(Object o) {
// write bundles size is the runner decision, we cannot force a bundle size,
// so we test the Writer as a DoFn outside of a runner.
- try (DoFnTester<Document, Document> fnTester =
- DoFnTester.of(new BulkIO.BulkIOBundleFn(write.getBulkIO()))) {
- // inserts into Elasticsearch
- fnTester.processBundle(serializeDocs(write, input));
- }
+ // try (DoFnTester<KV<Integer, Iterable<Document>>, Document> fnTester =
+ // DoFnTester.of(new BulkIOFn(write.getBulkIO()))) {
+ // // inserts into Elasticsearch
+ // fnTester.processBundle(serializeDocs(write, input));
+ // }
Review comment:
Can we remove the old code?
##########
File path: sdks/java/io/elasticsearch/src/main/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIO.java
##########
@@ -2335,126 +2270,40 @@ public void setup() throws IOException {
}
}
- @StartBundle
- public void startBundle(StartBundleContext context) {
- batch = ArrayListMultimap.create();
- currentBatchSizeBytes = 0;
- }
-
- /**
- * Adapter interface which provides a common parent for {@link ProcessContext} and {@link
- * FinishBundleContext} so that we are able to use a single common invocation to output from.
- */
- interface ContextAdapter {
- void output(
- TupleTag<Document> tag, Document document, Instant timestamp, BoundedWindow window);
- }
-
- private static final class ProcessContextAdapter<T> implements ContextAdapter {
- private final DoFn<T, Document>.ProcessContext context;
-
- private ProcessContextAdapter(DoFn<T, Document>.ProcessContext context) {
- this.context = context;
- }
-
- @Override
- public void output(
- TupleTag<Document> tag, Document document, Instant ignored1, BoundedWindow ignored2) {
- // Note: window and timestamp are intentionally unused, but required as params to fit the
- // interface
- context.output(tag, document);
- }
- }
-
- private static final class FinishBundleContextAdapter<T> implements ContextAdapter {
- private final DoFn<T, Document>.FinishBundleContext context;
-
- private FinishBundleContextAdapter(DoFn<T, Document>.FinishBundleContext context) {
- this.context = context;
- }
-
- @Override
- public void output(
- TupleTag<Document> tag, Document document, Instant timestamp, BoundedWindow window) {
- context.output(tag, document, timestamp, window);
+ @ProcessElement
+ public void processElement(ProcessContext c) throws IOException, InterruptedException {
+ if (c.element() == null || c.element().getValue() == null) {
+ return;
}
- }
-
- @FinishBundle
- public void finishBundle(FinishBundleContext context)
- throws IOException, InterruptedException {
- flushAndOutputResults(new FinishBundleContextAdapter<>(context));
- }
-
- private void flushAndOutputResults(ContextAdapter context)
- throws IOException, InterruptedException {
- // TODO: remove ContextAdapter and Multimap in favour of MultiOutputReceiver when
- // https://issues.apache.org/jira/browse/BEAM-1287 is completed
- Multimap<BoundedWindow, Document> results = flushBatch();
- for (Entry<BoundedWindow, Document> result : results.entries()) {
- BoundedWindow outputWindow = result.getKey();
- Document outputResult = result.getValue();
- Instant timestamp = outputResult.getTimestamp();
- if (timestamp == null) {
- timestamp = outputWindow.maxTimestamp();
- }
- if (outputResult.getHasError()) {
- context.output(Write.FAILED_WRITES, outputResult, timestamp, outputWindow);
+ for (Document doc : flushBatch(Lists.newArrayList(c.element().getValue()))) {
+ if (doc.getHasError()) {
+ c.output(Write.FAILED_WRITES, doc);
} else {
- context.output(Write.SUCCESSFUL_WRITES, outputResult, timestamp, outputWindow);
+ c.output(Write.SUCCESSFUL_WRITES, doc);
}
}
}
- protected void addAndMaybeFlush(
- Document bulkApiEntity, ProcessContext context, BoundedWindow outputWindow)
- throws IOException, InterruptedException {
-
- batch.put(outputWindow, bulkApiEntity);
- currentBatchSizeBytes +=
- bulkApiEntity.getBulkDirective().getBytes(StandardCharsets.UTF_8).length;
-
- if (batch.size() >= spec.getMaxBatchSize()
- || currentBatchSizeBytes >= spec.getMaxBatchSizeBytes()) {
- flushAndOutputResults(new ProcessContextAdapter<>(context));
- }
- }
-
- private boolean isRetryableClientException(Throwable t) {
- // RestClient#performRequest only throws wrapped IOException so we must inspect the
- // exception cause to determine if the exception is likely transient i.e. retryable or
- // not.
- return t.getCause() instanceof ConnectTimeoutException
- || t.getCause() instanceof SocketTimeoutException
- || t.getCause() instanceof ConnectionClosedException
- || t.getCause() instanceof ConnectException;
- }
-
- private Multimap<BoundedWindow, Document> flushBatch()
+ private Iterable<Document> flushBatch(List<Document> docs)
throws IOException, InterruptedException {
- if (batch.isEmpty()) {
- return ArrayListMultimap.create();
+ if (docs.isEmpty()) {
+ return Collections.emptyList();
}
- LOG.info(
- "ElasticsearchIO batch size: {}, batch size bytes: {}",
- batch.size(),
- currentBatchSizeBytes);
+ // LOG.info(
+ // "ElasticsearchIO batch size: {}, batch size bytes: {}",
+ // batch.size(),
+ // currentBatchSizeBytes);
Review comment:
Can we remove this code? Or do we want to keep it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] egalpin commented on pull request #17097: [BEAM-14064] ElasticsearchIO remove bundle based
Posted by GitBox <gi...@apache.org>.
egalpin commented on pull request #17097:
URL: https://github.com/apache/beam/pull/17097#issuecomment-1069323507
The prior unit test run took over 2 hours, and had a smattering of what look to be network failures. Retrying.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] egalpin commented on pull request #17097: [BEAM-14064] ElasticsearchIO remove bundle based
Posted by GitBox <gi...@apache.org>.
egalpin commented on pull request #17097:
URL: https://github.com/apache/beam/pull/17097#issuecomment-1071328286
Closing in favour of https://github.com/apache/beam/pull/17112
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] egalpin closed pull request #17097: [BEAM-14064] ElasticsearchIO remove bundle based
Posted by GitBox <gi...@apache.org>.
egalpin closed pull request #17097:
URL: https://github.com/apache/beam/pull/17097
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] egalpin commented on a change in pull request #17097: [BEAM-14064] ElasticsearchIO remove bundle based
Posted by GitBox <gi...@apache.org>.
egalpin commented on a change in pull request #17097:
URL: https://github.com/apache/beam/pull/17097#discussion_r828029641
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupIntoBatches.java
##########
@@ -221,6 +221,31 @@ private GroupIntoBatches(BatchingParams<InputT> params) {
duration));
}
+ public GroupIntoBatches<K, InputT> withByteSize(Long batchSizeBytes) {
+ checkArgument(
+ batchSizeBytes != null && batchSizeBytes < Long.MAX_VALUE && batchSizeBytes > 0,
+ "batchSizeBytes should be a non-negative value less than " + Long.MAX_VALUE);
+ return new GroupIntoBatches<>(
+ BatchingParams.create(
+ params.getBatchSize(),
+ batchSizeBytes,
+ params.getElementByteSize(),
+ params.getMaxBufferingDuration()));
+ }
+
+ public GroupIntoBatches<K, InputT> withByteSize(
Review comment:
cc: @reuvenlax since you seem to have made many of the recent changes to GroupIntoBatches. I added these methods to allow for users to make use of multiple batching criteria simultaneously. Another option for exposing this configuration option would be to implement a method like:
```java
public static <K, InputT> GroupIntoBatches<K, InputT> fromBatchingParams(BatchingParams params) {
return new GroupIntoBatches<>(params);
}
```
Thoughts on the preferred approach?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] egalpin commented on pull request #17097: [BEAM-14064] ElasticsearchIO remove bundle based
Posted by GitBox <gi...@apache.org>.
egalpin commented on pull request #17097:
URL: https://github.com/apache/beam/pull/17097#issuecomment-1069322834
Run Java PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] egalpin commented on pull request #17097: [BEAM-14064] ElasticsearchIO remove bundle based
Posted by GitBox <gi...@apache.org>.
egalpin commented on pull request #17097:
URL: https://github.com/apache/beam/pull/17097#issuecomment-1068530973
R: @lukecwik @je-ik @echauchot
Based on feedback from `Possible bug in ElasticsearchIO`[1]
[1] https://lists.apache.org/thread/mtwtno2o88lx3zl12jlz7o5w1lcgm2db
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] egalpin commented on a change in pull request #17097: [BEAM-14064] ElasticsearchIO remove bundle based
Posted by GitBox <gi...@apache.org>.
egalpin commented on a change in pull request #17097:
URL: https://github.com/apache/beam/pull/17097#discussion_r828025738
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupIntoBatches.java
##########
@@ -221,6 +221,31 @@ private GroupIntoBatches(BatchingParams<InputT> params) {
duration));
}
+ public GroupIntoBatches<K, InputT> withByteSize(Long batchSizeBytes) {
+ checkArgument(
+ batchSizeBytes != null && batchSizeBytes < Long.MAX_VALUE && batchSizeBytes > 0,
+ "batchSizeBytes should be a non-negative value less than " + Long.MAX_VALUE);
+ return new GroupIntoBatches<>(
+ BatchingParams.create(
+ params.getBatchSize(),
+ batchSizeBytes,
+ params.getElementByteSize(),
+ params.getMaxBufferingDuration()));
+ }
+
+ public GroupIntoBatches<K, InputT> withByteSize(
Review comment:
fixed 👍 thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] egalpin commented on pull request #17097: [BEAM-14064] ElasticsearchIO remove bundle based
Posted by GitBox <gi...@apache.org>.
egalpin commented on pull request #17097:
URL: https://github.com/apache/beam/pull/17097#issuecomment-1071096943
In light of the preference to continue supporting bundle-based batching in order to continue providing functionality across a wider variety of runners, I've created https://github.com/apache/beam/pull/17112.
I need some help determining whether the proposed solution of windowing inputs globally and then re-applying the input windowing strategy[1] is sufficient to ensure input windows are maintained.
[1] https://github.com/apache/beam/pull/17112/files#diff-e119923661d41dfc1911da8f6e29cc0e2e9a0a50129718c1c20c7d1aa32119bfR2273
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] egalpin commented on pull request #17097: [BEAM-14064] ElasticsearchIO remove bundle based
Posted by GitBox <gi...@apache.org>.
egalpin commented on pull request #17097:
URL: https://github.com/apache/beam/pull/17097#issuecomment-1071096943
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org