Trabajar con resultados paginados: análisis y consultas

Los métodos scan, query y batch de la API de cliente mejorado de DynamoDB devuelven respuestas con una o varias páginas. Una página contiene uno o varios elementos. El código puede procesar la respuesta por página o procesar elementos individuales.

Una respuesta paginada devuelta por el cliente síncrono DynamoDbEnhancedClient devuelve un objeto PageIterable, mientras que una respuesta devuelta por el asíncrono DynamoDbEnhancedAsyncClient devuelve un objeto PagePublisher.

En esta sección se analiza el procesamiento de los resultados paginados y se proporcionan ejemplos en los que se utilizan las API de análisis y consulta.

Examinar una tabla

El método scan del SDK corresponde a la operación de DynamoDB del mismo nombre. La API de cliente mejorado de DynamoDB ofrece las mismas opciones, pero utiliza un modelo de objetos conocido y gestiona la paginación por usted.

En primer lugar, exploramos la interfaz de PageIterable analizando el método scan de la clase de mapeo síncrono, DynamodbTable.

Utilizar la API síncrona

El siguiente ejemplo muestra el método scan que usa una expresión para filtrar los elementos que se devuelven. El ProductCatalog es el objeto modelo que se mostró anteriormente.

La expresión de filtrado que aparece después de la línea de comentario 2 limita los artículos ProductCatalog devueltos a aquellos con un precio comprendido entre 8 y 80 euros, ambos incluidos.

En este ejemplo también se excluyen los valores isbn mediante el método attributesToProject que se muestra después de la línea de comentario 1.

Después de la línea de comentario 3, el objeto PageIterable, pagedResults, es devuelto por el método scan. El método stream de PageIterable devuelve un objeto java.util.Stream, que puede utilizar para procesar las páginas. En este ejemplo, se cuenta y se registra el número de páginas.

Empezando por la línea de comentarios 4, el ejemplo muestra dos variantes del acceso a los elementos de ProductCatalog. La versión que sigue a la línea de comentarios 4a recorre cada página y ordena y registra los elementos de cada página. La versión posterior a la línea de comentarios 4b omite la iteración de la página y accede a los elementos directamente.

La interfaz PageIterable ofrece varias formas de procesar los resultados debido a sus dos interfaces principales, java.lang.Iterable y SdkIterable. Iterable trae los métodos forEach, iterator y spliterator, y SdkIterable el método stream.

    public static void scanSync(DynamoDbTable<ProductCatalog> productCatalog) {

        Map<String, AttributeValue> expressionValues = Map.of(
                ":min_value", numberValue(8.00),
                ":max_value", numberValue(80.00));

        ScanEnhancedRequest request = ScanEnhancedRequest.builder()
                .consistentRead(true)
                // 1. the 'attributesToProject()' method allows you to specify which values you want returned.
                .attributesToProject("id", "title", "authors", "price")
                // 2. Filter expression limits the items returned that match the provided criteria.
                .filterExpression(Expression.builder()
                        .expression("price >= :min_value AND price <= :max_value")
                        .expressionValues(expressionValues)
                        .build())
                .build();

        // 3. A PageIterable object is returned by the scan method.
        PageIterable<ProductCatalog> pagedResults = productCatalog.scan(request);
        logger.info("page count: {}", pagedResults.stream().count());

        // 4. Log the returned ProductCatalog items using two variations.
        // 4a. This version sorts and logs the items of each page.
        pagedResults.stream().forEach(p -> p.items().stream()
                .sorted(Comparator.comparing(ProductCatalog::price))
                .forEach(
                        item -> logger.info(item.toString())
                ));
        // 4b. This version sorts and logs all items for all pages.
        pagedResults.items().stream()
                .sorted(Comparator.comparing(ProductCatalog::price))
                .forEach(
                        item -> logger.info(item.toString())
                );
    }

Utilizar la API asíncrona

El método scan asíncrono devuelve los resultados como un objeto PagePublisher. La interfaz de PagePublisher tiene dos métodos subscribe que puede utilizar para procesar las páginas de respuesta. Un método subscribe proviene de la interfaz principal org.reactivestreams.Publisher. Para procesar páginas con esta primera opción, pase una instancia Subscriber al método subscribe. En el ejemplo siguiente se muestra el uso del método subscribe.

El segundo método subscribe proviene de la interfaz SDKPublisher. Esta versión de subscribe acepta un Consumer en lugar de un Subscriber. Esta variación del método subscribe se muestra en el segundo ejemplo siguiente.

El ejemplo siguiente muestra la versión asíncrona del método scan que utiliza la misma expresión de filtro que se muestra en el ejemplo anterior.

Tras la línea de comentario 3, DynamoDbAsyncTable.scan devuelve un objeto PagePublisher. En la siguiente línea, el código crea una instancia de la interfaz de org.reactivestreams.Subscriber, ProductCatalogSubscriber, que se suscribe a la PagePublisher después de la línea de comentarios 4.

El objeto Subscriber recopila los elementos ProductCatalog de cada página del método onNext después de la línea de comentarios 8 del ejemplo de clase ProductCatalogSubscriber. Los elementos se almacenan en la variable List privada y se accede a ellos en el código de llamada con el método ProductCatalogSubscriber.getSubscribedItems(). Esto se invoca después de la línea de comentarios 5.

Una vez recuperada la lista, el código ordena todos los artículos ProductCatalog por precio y registra cada uno de ellos.

El CountDownLatch de la clase ProductCatalogSubscriber bloquea el hilo de llamada hasta que todos los elementos se hayan agregado a la lista antes de continuar después de la línea de comentarios 5.

    public static void scanAsync(DynamoDbAsyncTable productCatalog) {
        ScanEnhancedRequest request = ScanEnhancedRequest.builder()
                .consistentRead(true)
                .attributesToProject("id", "title", "authors", "price")
                .filterExpression(Expression.builder()
                        // 1. :min_value and :max_value are placeholders for the values provided by the map
                        .expression("price >= :min_value AND price <= :max_value")
                        // 2. Two values are needed for the expression and each is supplied as a map entry.
                        .expressionValues(
                                Map.of( ":min_value", numberValue(8.00),
                                        ":max_value", numberValue(400_000.00)))
                        .build())
                .build();

        // 3. A PagePublisher object is returned by the scan method.
        PagePublisher<ProductCatalog> pagePublisher = productCatalog.scan(request);
        ProductCatalogSubscriber subscriber = new ProductCatalogSubscriber();
        // 4. Subscribe the ProductCatalogSubscriber to the PagePublisher.
        pagePublisher.subscribe(subscriber);
        // 5. Retrieve all collected ProductCatalog items accumulated by the subscriber.
        subscriber.getSubscribedItems().stream()
                .sorted(Comparator.comparing(ProductCatalog::price))
                .forEach(item ->
                        logger.info(item.toString()));
        // 6. Use a Consumer to work through each page.
        pagePublisher.subscribe(page -> page
                        .items().stream()
                        .sorted(Comparator.comparing(ProductCatalog::price))
                        .forEach(item ->
                                logger.info(item.toString())))
                .join(); // If needed, blocks the subscribe() method thread until it is finished processing.
        // 7. Use a Consumer to work through each ProductCatalog item.
        pagePublisher.items()
                .subscribe(product -> logger.info(product.toString()))
                .exceptionally(failure -> {
                    logger.error("ERROR  - ", failure);
                    return null;
                })
                .join(); // If needed, blocks the subscribe() method thread until it is finished processing.
    }

    private static class ProductCatalogSubscriber implements Subscriber<Page<ProductCatalog>> {
        private CountDownLatch latch = new CountDownLatch(1);
        private Subscription subscription;
        private List<ProductCatalog> itemsFromAllPages = new ArrayList<>();

        @Override
        public void onSubscribe(Subscription sub) {
            subscription = sub;
            subscription.request(1L);
            try {
                latch.await(); // Called by main thread blocking it until latch is released.
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }

        @Override
        public void onNext(Page<ProductCatalog> productCatalogPage) {
            // 8. Collect all the ProductCatalog instances in the page, then ask the publisher for one more page.
            itemsFromAllPages.addAll(productCatalogPage.items());
            subscription.request(1L);
        }

        @Override
        public void onError(Throwable throwable) {
        }

        @Override
        public void onComplete() {
            latch.countDown(); // Call by subscription thread; latch releases.
        }

        List<ProductCatalog> getSubscribedItems() {
            return this.itemsFromAllPages;
        }
    }

En el siguiente ejemplo de fragmento, se utiliza la versión del método PagePublisher.subscribe que acepta una Consumer después de la línea de comentario 6. El parámetro lambda de Java consume páginas, que procesan aún más cada elemento. En este ejemplo, se procesa cada página y los elementos de cada página se ordenan y, a continuación, se registran.

        // 6. Use a Consumer to work through each page.
        pagePublisher.subscribe(page -> page
                        .items().stream()
                        .sorted(Comparator.comparing(ProductCatalog::price))
                        .forEach(item ->
                                logger.info(item.toString())))
                .join(); // If needed, blocks the subscribe() method thread until it is finished processing.

El método items de PagePublisher separa las instancias del modelo para que el código pueda procesar los elementos directamente. Este método se muestra en el fragmento de código siguiente.

        // 7. Use a Consumer to work through each ProductCatalog item.
        pagePublisher.items()
                .subscribe(product -> logger.info(product.toString()))
                .exceptionally(failure -> {
                    logger.error("ERROR  - ", failure);
                    return null;
                })
                .join(); // If needed, blocks the subscribe() method thread until it is finished processing.

Consultar una tabla

Puede utilizar DynamoDB Enhanced Client para consultar la tabla y recuperar varios elementos que coincidan con criterios específicos. El método query() busca los elementos en función de los valores clave principales mediante la anotación @DynamoDbPartitionKey y, opcionalmente, la @DynamoDbSortKey definidas en la clase de datos.

El método query() requiere un valor de clave de partición y, de forma opcional, acepta condiciones de clave de clasificación para afinar aún más los resultados. Al igual que la API scan, las consultas devuelven una PageIterable para llamadas sincrónicas y una PagePublisher para asincrónicas.

Ejemplos del método Query

El ejemplo de código del método query() siguiente utiliza la clase MovieActor. La clase de datos define una clave primaria compuesta que se compone del atributo movie como clave de partición y el atributo actor como clave de clasificación.

package org.example.tests.model;

import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbAttribute;
import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbBean;
import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbPartitionKey;
import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbSecondaryPartitionKey;
import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbSecondarySortKey;
import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbSortKey;

import java.util.Objects;

@DynamoDbBean
public class MovieActor implements Comparable<MovieActor> {

    private String movieName;
    private String actorName;
    private String actingAward;
    private Integer actingYear;
    private String actingSchoolName;

    @DynamoDbPartitionKey
    @DynamoDbAttribute("movie")
    public String getMovieName() {
        return movieName;
    }

    public void setMovieName(String movieName) {
        this.movieName = movieName;
    }

    @DynamoDbSortKey
    @DynamoDbAttribute("actor")
    public String getActorName() {
        return actorName;
    }

    public void setActorName(String actorName) {
        this.actorName = actorName;
    }

    @DynamoDbSecondaryPartitionKey(indexNames = "acting_award_year")
    @DynamoDbAttribute("actingaward")
    public String getActingAward() {
        return actingAward;
    }

    public void setActingAward(String actingAward) {
        this.actingAward = actingAward;
    }

    @DynamoDbSecondarySortKey(indexNames = {"acting_award_year", "movie_year"})
    @DynamoDbAttribute("actingyear")
    public Integer getActingYear() {
        return actingYear;
    }

    public void setActingYear(Integer actingYear) {
        this.actingYear = actingYear;
    }

    @DynamoDbAttribute("actingschoolname")
    public String getActingSchoolName() {
        return actingSchoolName;
    }

    public void setActingSchoolName(String actingSchoolName) {
        this.actingSchoolName = actingSchoolName;
    }

    @Override
    public String toString() {
        final StringBuffer sb = new StringBuffer("MovieActor{");
        sb.append("movieName='").append(movieName).append('\'');
        sb.append(", actorName='").append(actorName).append('\'');
        sb.append(", actingAward='").append(actingAward).append('\'');
        sb.append(", actingYear=").append(actingYear);
        sb.append(", actingSchoolName='").append(actingSchoolName).append('\'');
        sb.append('}');
        return sb.toString();
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        MovieActor that = (MovieActor) o;
        return Objects.equals(movieName, that.movieName) && Objects.equals(actorName, that.actorName) && Objects.equals(actingAward, that.actingAward) && Objects.equals(actingYear, that.actingYear) && Objects.equals(actingSchoolName, that.actingSchoolName);
    }

    @Override
    public int hashCode() {
        return Objects.hash(movieName, actorName, actingAward, actingYear, actingSchoolName);
    }

    @Override
    public int compareTo(MovieActor o) {
        if (this.movieName.compareTo(o.movieName) != 0){
            return this.movieName.compareTo(o.movieName);
        } else {
            return this.actorName.compareTo(o.actorName);
        }
    }
}

Los ejemplos de código que aparecen a continuación se refieren a los siguientes elementos.

MovieActor{movieName='movie01', actorName='actor0', actingAward='actingaward0', actingYear=2001, actingSchoolName='null'}
MovieActor{movieName='movie01', actorName='actor1', actingAward='actingaward1', actingYear=2001, actingSchoolName='actingschool1'}
MovieActor{movieName='movie01', actorName='actor2', actingAward='actingaward2', actingYear=2001, actingSchoolName='actingschool2'}
MovieActor{movieName='movie01', actorName='actor3', actingAward='actingaward3', actingYear=2001, actingSchoolName='null'}
MovieActor{movieName='movie01', actorName='actor4', actingAward='actingaward4', actingYear=2001, actingSchoolName='actingschool4'}
MovieActor{movieName='movie02', actorName='actor0', actingAward='actingaward0', actingYear=2002, actingSchoolName='null'}
MovieActor{movieName='movie02', actorName='actor1', actingAward='actingaward1', actingYear=2002, actingSchoolName='actingschool1'}
MovieActor{movieName='movie02', actorName='actor2', actingAward='actingaward2', actingYear=2002, actingSchoolName='actingschool2'}
MovieActor{movieName='movie02', actorName='actor3', actingAward='actingaward3', actingYear=2002, actingSchoolName='null'}
MovieActor{movieName='movie02', actorName='actor4', actingAward='actingaward4', actingYear=2002, actingSchoolName='actingschool4'}
MovieActor{movieName='movie03', actorName='actor0', actingAward='actingaward0', actingYear=2003, actingSchoolName='null'}
MovieActor{movieName='movie03', actorName='actor1', actingAward='actingaward1', actingYear=2003, actingSchoolName='actingschool1'}
MovieActor{movieName='movie03', actorName='actor2', actingAward='actingaward2', actingYear=2003, actingSchoolName='actingschool2'}
MovieActor{movieName='movie03', actorName='actor3', actingAward='actingaward3', actingYear=2003, actingSchoolName='null'}
MovieActor{movieName='movie03', actorName='actor4', actingAward='actingaward4', actingYear=2003, actingSchoolName='actingschool4'}

El siguiente código define dos instancias QueryConditional: keyEqual (después de la línea de comentario 1) y sortGreaterThanOrEqualTo (después de la línea de comentario 1a).

Consulta de elementos por clave de partición

La instancia keyEqual asocia los elementos con un valor de clave de partición de movie01.

Este ejemplo también define una expresión de filtro después de la línea de comentario 2 que filtra cualquier elemento que no tenga un valor de actingschoolname.

La QueryEnhancedRequest combina la condición de clave y la expresión de filtro de la consulta.

    public static void query(DynamoDbTable movieActorTable) {

        // 1. Define a QueryConditional instance to return items matching a partition value.
        QueryConditional keyEqual = QueryConditional.keyEqualTo(b -> b.partitionValue("movie01"));
        // 1a. Define a QueryConditional that adds a sort key criteria to the partition value criteria.
        QueryConditional sortGreaterThanOrEqualTo = QueryConditional.sortGreaterThanOrEqualTo(b -> b.partitionValue("movie01").sortValue("actor2"));
        // 2. Define a filter expression that filters out items whose attribute value is null.
        final Expression filterOutNoActingschoolname = Expression.builder().expression("attribute_exists(actingschoolname)").build();

        // 3. Build the query request.
        QueryEnhancedRequest tableQuery = QueryEnhancedRequest.builder()
                .queryConditional(keyEqual)
                .filterExpression(filterOutNoActingschoolname)
                .build();
        // 4. Perform the query using the "keyEqual" conditional and filter expression.
        PageIterable<MovieActor> pagedResults = movieActorTable.query(tableQuery);
        logger.info("page count: {}", pagedResults.stream().count()); // Log  number of pages.

        pagedResults.items().stream()
                .sorted()
                .forEach(
                        item -> logger.info(item.toString()) // Log the sorted list of items.
                );

2023-03-05 13:11:05 [main] INFO  org.example.tests.QueryDemo:46 - page count: 1
2023-03-05 13:11:05 [main] INFO  org.example.tests.QueryDemo:51 - MovieActor{movieName='movie01', actorName='actor1', actingAward='actingaward1', actingYear=2001, actingSchoolName='actingschool1'}
2023-03-05 13:11:05 [main] INFO  org.example.tests.QueryDemo:51 - MovieActor{movieName='movie01', actorName='actor2', actingAward='actingaward2', actingYear=2001, actingSchoolName='actingschool2'}
2023-03-05 13:11:05 [main] INFO  org.example.tests.QueryDemo:51 - MovieActor{movieName='movie01', actorName='actor4', actingAward='actingaward4', actingYear=2001, actingSchoolName='actingschool4'}

Consulta de elementos por clave de partición y clave de clasificación

La QueryConditional sortGreaterThanOrEqualTo afina la coincidencia de clave de partición (movie01) añadiendo una condición de clave de clasificación para valores mayores o iguales que actor2.

Los métodos de QueryConditional que comienzan por sort requieren un valor de clave de partición para asociar y afinar aún más la consulta mediante una comparación basada en el valor de la clave de clasificación. Sort en el nombre del método no significa que los resultados estén clasificados, sino que se utilizará un valor de clave de clasificación para comparación.

En el siguiente fragmento cambiamos la solicitud de consulta que mostrada anteriormente después de la línea de comentario 3. Este fragmento reemplaza el condicional de consulta “KeyEqual” por el condicional de consulta “SortGreaterThanOrEqualTo” que se ha definido después de la línea de comentario 1a. El código siguiente también elimina la expresión del filtro.

        QueryEnhancedRequest tableQuery = QueryEnhancedRequest.builder()
                .queryConditional(sortGreaterThanOrEqualTo).build();

2023-03-05 13:15:00 [main] INFO  org.example.tests.QueryDemo:46 - page count: 1
2023-03-05 13:15:00 [main] INFO  org.example.tests.QueryDemo:51 - MovieActor{movieName='movie01', actorName='actor2', actingAward='actingaward2', actingYear=2001, actingSchoolName='actingschool2'}
2023-03-05 13:15:00 [main] INFO  org.example.tests.QueryDemo:51 - MovieActor{movieName='movie01', actorName='actor3', actingAward='actingaward3', actingYear=2001, actingSchoolName='null'}
2023-03-05 13:15:00 [main] INFO  org.example.tests.QueryDemo:51 - MovieActor{movieName='movie01', actorName='actor4', actingAward='actingaward4', actingYear=2001, actingSchoolName='actingschool4'}