[ https://issues.apache.org/jira/browse/BEAM-6886?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16801489#comment-16801489 ]
Etienne Chauchot commented on BEAM-6886: ---------------------------------------- [~MadEgg] You are referring to "batch size" but you are forgetting the most important part of the name of the method which is MAX. It is MAXBatchSize. ElasticsearchIO cannot control the size of the batch because it is dependent of the bundle size which is the responsibility of the runner to set. Depending on what the runner targets it sets the bundle size accordingly: a streaming job (as you supposed correctly) tends to create small bundles to lower latency and increased throughput, on the contrary a runner that targets batch jobs, tends to create higher size bundles. For example Flink creates one element bundles so if you used this runner, you'd get 1 element ES bulks. Regarding the javadoc, it is already said in the withMaxBatchSize() javadoc, quote: "Depending on the execution engine, size of bundles may vary, this sets the maximum size" Regarding your proposal of outputting only when the window closes would be against beam model (violates the bundle unit of commit) and would lead in batch mode to output only at the end of the batch, as, in that mode, all the elements are stored in a single window called Global window. Regarding your workaround: IMHO GroupIntoBatches is the correct way to go as said by [~kenn]. Having to key your PCollection is needed only if you do not have a keyed PCollection already (which I suppose you don't). It is because the GIB uses state API which is per key and window (otherwise we would have a huge state and we could not clean in when a given key is done). IMHO including GIB into ESIO seems less composable than allowing the user to chose whether he needs it and moreover it would need to be done in all IOs. Nevertheless, what can be done to reduce the awkwardness of the pipeline you mention is detecting in GIB that the PCollection is not keyed and key it at the beginning and unkey it at the end. [~kenn] [~MadEgg] WDYT ? > Change batch handling in ElasticsearchIO to avoid necessity for > GroupIntoBatches > -------------------------------------------------------------------------------- > > Key: BEAM-6886 > URL: https://issues.apache.org/jira/browse/BEAM-6886 > Project: Beam > Issue Type: Improvement > Components: io-java-elasticsearch > Affects Versions: 2.11.0 > Reporter: Egbert > Priority: Major > > I have a streaming job inserting records into an Elasticsearch cluster. I set > the batch size appropriately big, but I found out this is not causing any > effect at all: I found that all elements are inserted in batches of 1 or 2 > elements. > The reason seems to be that this is a streaming pipeline, which may result in > tiny bundles. Since ElasticsearchIO uses `@FinishBundle` to flush a batch, > this will result in equally small batches. > This results in a huge amount of bulk requests with just one element, > grinding the Elasticsearch cluster to a halt. > I have now been able to work around this by using a `GroupIntoBatches` > operation before the insert, but this results in 3 steps (mapping to a key, > applying GroupIntoBatches, stripping key and outputting all collected > elements), making the process quite awkward. > A much better approach would be to internalize this into the ElasticsearchIO > write transform.. Use a timer that flushes the batch at batch size or end of > window, not at the end of a bundle. -- This message was sent by Atlassian JIRA (v7.6.3#76005)