Batch processing with reactive Couchbase java driver

user_3128170 0
Viewed 184

Suppose I have a bucket from which I need to fetch documents that have a date older than now. This document looks like this:

{
id: "1",
date: "Some date",
otherObjectKEY: "key1"
}

For each document, I need to fetch another document using its otherObjectKEY, send the latter to a kafka topic, then delete the original document.

Using the reactive java driver 3.0, I was able to do it with something like this:

public void batch(){
    streamOriginalObjects()
         .flatMap(originalObject -> fetchOtherObjectUsingItsKEY(originalObject)
                       .flatMap(otherObject -> sendToKafkaAndDeleteOriginalObject(originalObject))
         )
         .subscribe();
}

streamOriginalObjects():

public Flux<OriginalObject> streamOriginalObjects(){
        return client.query("select ... and date <= '"+ LocalDateTime.now().toString() +"'")
                .flux()
                .flatMap(result -> result.rowsAs(OriginalObject.class));
    }

It works like expected, but I'm wondering if there is a better approach (especially in terms of performance) than streaming and processing element by element.

couchbase reactive-programming
1 Answers

Doing a N1QL query, and then fanning-out key-value operations from that, is a useful and common pattern. This should make the fan-out happen in parallel:

    streamOriginalObjects()
        // Split into numberOfThreads 'rails'
        .parallel(numberOfThreads)

        // Run on an unlimited thread pool
        .runOn(Schedulers.elastic())

        .concatMap(originalObject -> fetchOtherObjectUsingItsKEY(originalObject)
            .concatMap(otherObject -> sendToKafkaAndDeleteOriginalObject(originalObject))
        )

        // Back out of parallel mode
        .sequential()
        .subscribe();
user_46881150
Related