[BEAM-12093] Overhaul ElasticsearchIO.Write

[egalpin]

This change set represents a rather large (and backward compatible) change to the way ElastichsearchIO.Write operates. Presently, the Write transform has 2 responsibilities:

1. Convert input documents into Bulk API entities, serializing based on user settings (partial update, delete, upsert, etc) -> DocToBulk
2. Batch the converted Bulk API entities together and interface with the target ES cluster -> BulkIO

This PR aims to separate these 2 responsibilities into discrete PTransforms to allow for greater flexibility while also maintaining the convenience of the Write transform to perform both document conversion and IO serially. Examples of how the flexibility of separate transforms could be used:

1. Unit testing. It becomes trivial for pipeline developers to ensure that output Bulk API entities for a given set of inputs will produce an expected result, without the need for an available Elasticsearch cluster.
2. Flexible options for data backup. Serialized Bulk API entities can be forked and sent to both Elasticsearch and a data lake.
3. Mirroring data to multiple clusters. Presently, mirroring data to multiple clusters would require duplicate computation.
4. Better batching with input streams in one job. A job may produce multiple "shapes" of Bulk API entities based on multiple input types, and then "fan-in" all serialized Bulk entities into a single BulkIO transform to improve batching semantics.
5. Decoupled jobs. Corollary to (4) above. Job(s) could be made to produce Bulk entities and then publish them to a message bus. A distinct job could consume from that message bus and solely be responsible for IO with the target cluster(s).
6. Easier support for multiple BulkIO semantics.

Expanding on point (6), this PR also introduces a new (optional) way to batch entities for bulk requests: Stateful Processing. Presently, Bulk request size is limited by the lesser of Runner bundle size and maxBatchSize user setting. In my experience, bundle sizes are often very small, and can be a small as 1 or 2. When that’s the case, it means Bulk requests contain only 1 or 2 documents, and it’s effectively the same as not using the Bulk API at all. BulkIOStatefulFn is made to be compatible with GroupIntoBatches which will use entity count and (optionally) elapsed time to create batches much closer to the maxBatchSize setting to improve throughput.

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Choose reviewer(s) 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.

See the Contributor Guide for more tips on how to make review process smoother.

Post-Commit Tests Status (on master branch)

Lang	SDK	ULR	Dataflow	Flink	Samza	Spark	Twister2
Go		---	---		---		---
Java
Python		---			---		---
XLang		---			---		---

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website	Whitespace	Typescript
Non-portable
Portable	---		---	---	---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

[egalpin]

As this is my first time contributing, I'm not sure exactly who to select as a reviewer. I see @echauchot in the git blame a lot for ElasticsearchIO.java, so I'll start there? 😂

[echauchot]

Sure, I'll be happy to review. Thanks for your contribution !

[egalpin]

I have the local dev env setup by using start-build-env.sh from this repo, but I'm still working towards running elasticsearch-tests. Any pointers would be appreciated if anyone has time 🙂

[echauchot]

@egalpin thanks for your contribution. I'm sorry I lack time a lot lately. In the meantime can you:

• make the build pass (precommit fail)
• as it is rather large contribution : please open a ticket with details and rename PR name.

[egalpin]

@echauchot I've made a jira ticket and linked it. I'm working on getting the build to pass but struggling a bit with trying to determine the cause of the new errors in the Java PreCommit build. All the warnings have to do with a single Kotlin example which seems far removed from the changes here. I'll keep poking away at it though. I've found the full output from the build and can see some compilation errors. Working on those.

[egalpin]

@echauchot FYI build is passing now 🙂

[pabloem]

@echauchot let me know if you can take a look at this, and if not I can help you find more reviewers : )

[egalpin]

@pabloem fyi I messaged echauchot on Apache beam slack previously and we had arranged to chat about tests together tomorrow. I wanted to let them know that I’d gotten past my blockers with respect to tests, but I definitely did not intend to apply pressure or name+shame or anything like that.

[pabloem]

oh I also didn't want to pressure anyone : ) I'm just checking. thanks for the update!

[egalpin]

Haha thanks! I appreciate it. I wanted to be sure I was respecting others' busy schedules ❤️

[echauchot]

@egalpin starting first round of review sorry for the delay

[echauchot]

@timrobertson100 you might be interested. @ludovic-boutros you were thinking about changing the overall arhictecture of the IO and introduce a testContainer based test framework here: https://lists.apache.org/thread.html/reb68f37c435995a64ded19100e09dfc31c5cf6227feae16494226100%40%3Cdev.beam.apache.org%3E
any comment ?

[echauchot]

@echauchot let me know if you can take a look at this, and if not I can help you find more reviewers : )

@pabloem  @egalpin I'll do an overall review but I'd need another reviewer to do the in-depth review because:

• I'm busy on many other things lately
• I coded support for ES 2, ES 5, reviewed 6 and coded ES7. I'd like to pass the torch :)

[egalpin]

Thanks @echauchot for sharing that thread. I really like a lot of what @ludovic-boutros proposed and have many shared goals; in particular, implementing the pattern where successful and failed writes can be returned via MultiOutputReceiver. I believe the multi-output pattern could fit within the current IO with some additional effort (since order of request and response entities is guaranteed[1]), and I had planned to do that as a follow-up so as to not introduce even more changes in one PR.

At the same time, I also see the argument that in many ways using the low-level client results in "reinventing the wheel" for a number of features (with good justification, IMO, of enabling cross-version support).

I'd be very willing to contribute to brainstorming (and implementation once we reach that point) if others are open to that.

[1] https://discuss.elastic.co/t/order-of-actions-in-bulk-api-via-http-between-request-and-response-is-guaranteed/122499/2

[egalpin]

Run Java PreCommit

[ludovic-boutros]

@egalpin @echauchot I made a quick review, but, well, sadly I don't have enough time these months to go deeper on the subject.
We are using the module I shared in production for months without any issue.
The development of a new multi module component is currently in stand by with one module per Elasticsearch version and a top level abstraction.
It's not finished yet and with the Covid situation I had to refocus on other projects. That means that I will not get any time to really go further soon. I'm available to share on this with you (Slack ?).
I think political decision should be taken first before taking the direction I proposed. But it's not the place to have this discussion.

[echauchot]

This change set represents a rather large (and backward compatible) change to the way ElastichsearchIO.Write operates. Presently, the Write transform has 2 responsibilities:

1. Convert input documents into Bulk API entities, serializing based on user settings (partial update, delete, upsert, etc) -> DocToBulk
2. Batch the converted Bulk API entities together and interface with the target ES cluster -> BulkIO

This PR aims to separate these 2 responsibilities into discrete PTransforms to allow for greater flexibility while also maintaining the convenience of the Write transform to perform both document conversion and IO serially. Examples of how the flexibility of separate transforms could be used:

1. Unit testing. It becomes trivial for pipeline developers to ensure that output Bulk API entities for a given set of inputs will produce an expected result, without the need for an available Elasticsearch cluster.
2. Flexible options for data backup. Serialized Bulk API entities can be forked and sent to both Elasticsearch and a data lake.
3. Mirroring data to multiple clusters. Presently, mirroring data to multiple clusters would require duplicate computation.
4. Better batching with input streams in one job. A job may produce multiple "shapes" of Bulk API entities based on multiple input types, and then "fan-in" all serialized Bulk entities into a single BulkIO transform to improve batching semantics.
5. Decoupled jobs. Corollary to (4) above. Job(s) could be made to produce Bulk entities and then publish them to a message bus. A distinct job could consume from that message bus and solely be responsible for IO with the target cluster(s).
6. Easier support for multiple BulkIO semantics.

=> Reading at the overall design goals, it looks very promising and a good analysis of the missing properties of the curent architecture ! Thanks !

Expanding on point (6), this PR also introduces a new (optional) way to batch entities for bulk requests: Stateful Processing. Presently, Bulk request size is limited by the lesser of Runner bundle size and maxBatchSize user setting. In my experience, bundle sizes are often very small, and can be a small as 1 or 2. When that’s the case, it means Bulk requests contain only 1 or 2 documents, and it’s effectively the same as not using the Bulk API at all. BulkIOStatefulFn is made to be compatible with GroupIntoBatches which will use entity count and (optionally) elapsed time to create batches much closer to the maxBatchSize setting to improve throughput.

=> True that very small batches can exist for example Flink being a streaming oriented platform, Flink runner tends to create very small Beam bundles. So, when the bundle is finished processing (finishBundle is called), the ES bulk request is sent leading to small ES bulk. Leveraging GroupIntoBatches that creates trans-bundle groups and still respect Beam semantics (windowing, bundle retries etc...) is a very good idea.

[echauchot]

Very good work Evan ! Thanks.

I like the analysis of the missing features and all the improvements you gave !
I did a pretty in depth review after all but I no more have time. @pabloem as you offered help, could you please do the other rounds of review and merge when ok ?

Besides, Evan, as you know ES very well, and you seem to be interested in contributing. Would you be interested in putting yourself in ES Owners file and jira ES label ?

[egalpin]

@echauchot Thanks for the review, I'll work my way through your comments and suggestions.

Besides, Evan, as you know ES very well, and you seem to be interested in contributing. Would you be interested in putting yourself in ES Owners file and jira ES label ?

I'd be very happy to 👍 I've added myself to the ES owners file now, happy to lend a hand reviewing! Thanks 🙂

With respect to Jira, could you please add appropriate permissions for me to either assign myself to the ES label, or assign me to the label yourself if that is the preferred workflow. I have an account on issues.apache.org/jira but only with permission to create tickets I believe.

[egalpin]

@echauchot added coverage for the methods you mentioned. Anything else outstanding? 🙂

[echauchot]

only minor changes left. I think when you address them, I could hit the merge button provided the tests pass.

[egalpin]

Run Java PreCommit

[egalpin]

Run Java PreCommit

[egalpin]

Run Java PreCommit

[egalpin]

Ready to roll! 🙂

[echauchot]

@egalpin seems good ! thanks ! I just triggered the build. it's like fort knox now on resources consumption 😄

[echauchot]

@egalpin ready to merge. only the 2 javadoc (assert numdocs/num scientits) fixes and the not needed gbk in parallel test (+consequent simplification of assertion function) and we're good to merge.

[egalpin]

@echauchot all set for a final look-over 🙂

[egalpin]

Run Java PreCommit

[egalpin]

Run Java PreCommit

[egalpin]

Run Java PreCommit

[echauchot]

LGTM,
Thanks for you great and hard work Evan ! And also thanks for your patience.
As stated in the guidelines, I'll squash the review commits into the first commit and merge.

[egalpin]

Thanks Etienne for all of your reviewing efforts, and your warm welcome to Beam! 😄

[mattwelke]

Came across this while lurking. Where would one find docs on how to use ElasticsearchIO, including more advanced features like this? I saw some examples on the Beam site, but nothing specific to each source/sink.

[egalpin]

@mattwelke The javadoc[1] that is generated has some description and examples, though admittedly not as fully descriptive as it could be. I would welcome any additional examples or documentation added by others, and will keep in mind to add more examples when time permits!

Do you have any specific questions or a use case you would like help determining how to best use this IO? I might suggest that we move the conversation to the user mailing list or slack[2] so that others could more easily benefit from our conversation 🙂

[1] https://beam.apache.org/releases/javadoc/2.35.0/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIO.html

[2] https://beam.apache.org/community/join-beam/

[timrobertson100]

The tests might also be worth looking at @mattwelke

E.g.
https://github.com/apache/beam/blob/master/sdks/java/io/elasticsearch-tests/elasticsearch-tests-common/src/test/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIOTestCommon.java#L770

[mattwelke]

Those are a good start, thanks. For any more discussion, I'll use the mailing list or Slack.