What is the recommended way to stream Elasticsearch index data

orange.cold256 · February 29, 2024, 10:18am

Good afternoon!

I have a single-node Elasticsearch cluster with a pair of indexes. I'm using Java Spring Boot backend that provides Elasticsearch queries via rest high level client, but I need to provide data streaming somehow.

The point is to provide user with a application/x-ndjson or application/stream+json stream to be able to see the real-time changes. When a new document is being created / modified I should be able to stream those changes in real time.

As far as I understand, all I need is to use reactive Elasticsearch client with reactive web client, like Webflux to pass the response.

But there are few questions that I can't find answers to:

Do I need to modify my current index structure to support reactive behaviour? 2. Are there any official recommendations for data streaming?
Would it be possible to handle long-term streams, that could last, for example, for 24 hours?

Would be really appreciated for some links

dadoonet · February 29, 2024, 4:46pm

I'd use definitely the new Java client. It has an async version which is IMO designed for that kind of things.

I guess that using the BulkIngester class would help a lot as well to batch a bit your requests.

Have a look at

github.com

dadoonet/elasticsearch-java-client-demo/blob/main/src/test/java/fr/pilato/test/elasticsearch/hlclient/EsClientIT.java#L407


      
              assertNotNull(response.aggregations().get("top10foo").sterms());
              assertNotNull(response.aggregations().get("top10foo").sterms().buckets());
              assertEquals(1, response.aggregations().get("top10foo").sterms().buckets().array().size());
          
              StringTermsBucket top10foo = response.aggregations().get("top10foo").sterms().buckets().array().get(0);
              assertEquals("bar", top10foo.key().stringValue());
              assertEquals(2, top10foo.docCount());
          }
          
          @Test
          void bulkIngester() throws IOException {
              int size = 1000;
              try (BulkIngester<Void> ingester = BulkIngester.of(b -> b
                      .client(client)
                      .listener(new BulkListener<>() {
                          @Override
                          public void beforeBulk(long executionId, BulkRequest request, List<Void> voids) {
                              logger.debug("going to execute bulk of {} requests", request.operations().size());
                          }
          
                          @Override

system · March 28, 2024, 4:47pm

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.