Construtores de agregados

Visão geral

Neste guia, veja como usar a classe Agregados que fornece métodos estáticos de fábrica para construir agregação pipeline stages no driver MongoDB Kotlin .

Para uma introdução mais completa à agregação, consulte nosso guia de agregação.

Os exemplos nesta página pressupõem importações para métodos das seguintes classes:

• Aggregates

• Filters

• Projections

• Sorts

• Accumulators

import com.mongodb.client.model.Aggregates
import com.mongodb.client.model.Filters
import com.mongodb.client.model.Projections
import com.mongodb.client.model.Sorts
import com.mongodb.client.model.Accumulators

Use esses métodos para criar estágios de pipeline e especificá-los em sua agregação como uma lista:

val matchStage = Aggregates.match(Filters.eq("someField", "someCriteria"))
val sortByCountStage = Aggregates.sortByCount("\$someField")
val results = collection.aggregate(
    listOf(matchStage, sortByCountStage)).toList()

Many Aggregation examples in this guide use the Atlas sample_mflix.movies dataset. The documents in this collection are modeled by the following Movie data class for use with the Kotlin driver:

data class Movie(
    val title: String,
    val year: Int,
    val genres: List<String>,
    val rated: String,
    val plot: String,
    val runtime: Int,
    val imdb: IMDB,
    val fullplot: String? = "No full plot",
){
    data class IMDB(
        val rating: Double
    )
}

corresponder

Use the match() method to create a $match pipeline stage that matches incoming documents against the specified query filter, filtering out documents that do not match.

O exemplo a seguir cria um estágio de pipeline que corresponde a todos os documento na collection movies em que o campo title é igual a "The Shawshank Redemption":

Aggregates.match(Filters.eq(Movie::title.name, "The Shawshank Redemption"))

Projeto

Use o método project() para criar um estágio de pipeline $project em que os campos do documento especificados pelo projeto. A projeção de campos em agregação segue as mesmas regras da projeção de campos em queries.

O exemplo a seguir cria um estágio de pipeline que inclui os campos title e plot , mas exclui o campo _id :

Aggregates.project(
    Projections.fields(
        Projections.include(Movie::title.name, Movie::plot.name),
        Projections.excludeId())
)

Aggregates.project(
    Projections.fields(
        Projections.computed("rating", "\$${Movie::rated.name}"),
        Projections.excludeId()
    )
)

Documentos

Use the documents() method to create a $documents pipeline stage that returns literal documents from input values.

O exemplo abaixo cria um estágio de pipeline que gera documentos de amostra na collection movies com um campo title :

Aggregates.documents(
    listOf(
        Document(Movie::title.name, "Steel Magnolias"),
        Document(Movie::title.name, "Back to the Future"),
        Document(Movie::title.name, "Jurassic Park")
    )
)

val docsStage = database.aggregate<Document>( // ... )

Amostra

Use o método sample() para criar um estágio de pipeline $sample para selecionar aleatoriamente documentos da entrada.

O exemplo abaixo cria um estágio de pipeline que seleciona aleatoriamente cinco documentos da coleção movies:

Aggregates.sample(5)

Sort

Use o método sort() para criar um estágio de pipeline $sort para classificar de acordo com os critérios especificados.

O exemplo a seguir cria um estágio de pipeline que classifica em ordem decrescente de acordo com o valor do campo year e, em seguida, em ordem crescente de acordo com o valor do campo title:

Aggregates.sort(
    Sorts.orderBy(
        Sorts.descending(Movie::year.name),
        Sorts.ascending(Movie::title.name)
    )
)

Ignorar

Use the skip() method to create a $skip pipeline stage to skip over the specified number of documents before passing documents into the next stage.

O exemplo a seguir cria um estágio de pipeline que ignora os primeiros documento 5 na collection movies :

Aggregates.skip(5)

Limite

Use o estágio do pipeline $limit para limitar o número de documentos passados para o próximo estágio.

O exemplo a seguir cria um estágio de pipeline que limita o número de documento retornados da collection movies para 4:

Aggregates.limit(4)

Pesquisa

Use the lookup() method to create a $lookup pipeline stage to perform joins and uncorrelated subqueries between two collections.

Aggregates.lookup(
    "comments",
    "_id",
    "movie_id",
    "joined_comments"
)

data class Order(
    @BsonId val id: Int,
    val customerId: Int,
    val item: String,
    val ordered: Int
)
data class Inventory(
    @BsonId val id: Int,
    val stockItem: String,
    val inStock: Int
)

val variables = listOf(
    Variable("order_item", "\$item"),
    Variable("order_qty", "\$ordered")
)
val pipeline = listOf(
    Aggregates.match(
        Filters.expr(
            Document("\$and", listOf(
                Document("\$eq", listOf("$\$order_item", "\$${Inventory::stockItem.name}")),
                Document("\$gte", listOf("\$${Inventory::inStock.name}", "$\$order_qty"))
            ))
        )
    ),
    Aggregates.project(
        Projections.fields(
            Projections.exclude(Order::customerId.name, Inventory::stockItem.name),
            Projections.excludeId()
        )
    )
)
val innerJoinLookup =
    Aggregates.lookup("warehouses", variables, pipeline, "stockData")

Grupo

Use the group() method to create a $group pipeline stage to group documents by a specified expression and output a document for each distinct grouping.

O exemplo a seguir cria um estágio de pipeline que agrupa documento na collection orders pelo valor do campo customerId . Cada grupo acumula a soma e média dos valores do campo ordered nos campos totalQuantity e averageQuantity :

Aggregates.group("\$${Order::customerId.name}",
    Accumulators.sum("totalQuantity", "\$${Order::ordered.name}"),
    Accumulators.avg("averageQuantity", "\$${Order::ordered.name}")
)

Learn more about accumulator operators from the Server manual section on Accumulators.

Pick-N Accumulators

Os acumuladores pick-n são operadores de acumulação de agregação que retornam os elementos superiores e inferiores de acordo com uma ordem específica. Use um dos seguintes construtores para criar um operador de acumulação de agregação:

• minN()

• maxN()

• firstN()

• lastN()

• top()

• topN()

• bottom()

• bottomN()

Saiba com quais aggregation pipeline stages você pode usar operadores acumuladores acessando a seção manual do servidor sobre Acumuladores.

Os exemplos do acumulador do pick-n usam documento da collection do movies no reconhecimento de data center do sample-mflix .

Aggregates.group(
    "\$${Movie::year.name}",
    Accumulators.minN(
        "lowestThreeRatings",
        "\$${Movie::imdb.name}.${Movie.IMDB::rating.name}",
        3
    )
)

Aggregates.group(
    "\$${Movie::year.name}",
    Accumulators.maxN(
        "highestTwoRatings",
        "\$${Movie::imdb.name}.${Movie.IMDB::rating.name}",
        2
    )
)

Aggregates.group(
    "\$${Movie::year.name}",
    Accumulators.firstN(
        "firstTwoMovies",
        "\$${Movie::title.name}",
        2
    )
)

Aggregates.group(
    "\$${Movie::year.name}",
    Accumulators.lastN(
        "lastThreeMovies",
        "\$${Movie::title.name}",
        3
    )
)

Aggregates.group(
    "\$${Movie::year.name}",
    Accumulators.top(
        "topRatedMovie",
        Sorts.descending("${Movie::imdb.name}.${Movie.IMDB::rating.name}"),
        listOf("\$${Movie::title.name}", "\$${Movie::imdb.name}.${Movie.IMDB::rating.name}")
    )
)

Aggregates.group(
    "\$${Movie::year.name}",
    Accumulators.topN(
        "longestThreeMovies",
        Sorts.descending(Movie::runtime.name),
        listOf("\$${Movie::title.name}", "\$${Movie::runtime.name}"),
        3
    )
)

Aggregates.group(
    "\$${Movie::year.name}",
    Accumulators.bottom(
        "shortestMovies",
        Sorts.descending(Movie::runtime.name),
        listOf("\$${Movie::title.name}", "\$${Movie::runtime.name}")
    )
)

Aggregates.group(
    "\$${Movie::year.name}",
    Accumulators.bottom(
        "lowestRatedTwoMovies",
        Sorts.descending("${Movie::imdb.name}.${Movie.IMDB::rating.name}"),
        listOf("\$${Movie::title.name}", "\$${Movie::imdb.name}.${Movie.IMDB::rating.name}"),
    )
)

Unwind

Use the unwind() method to create an $unwind pipeline stage to deconstruct an array field from input documents, creating an output document for each array element.

O exemplo a seguir cria um documento para cada elemento na array lowestRatedTwoMovies:

Aggregates.unwind("\$${"lowestRatedTwoMovies"}")

Para preservar documentos que têm valores ausentes ou null para o campo de array, ou onde a array está vazia:

Aggregates.unwind(
    "\$${"lowestRatedTwoMovies"}",
    UnwindOptions().preserveNullAndEmptyArrays(true)
)

Para incluir o índice da array (neste exemplo, em um campo chamado "position"):

Aggregates.unwind(
    "\$${"lowestRatedTwoMovies"}",
    UnwindOptions().includeArrayIndex("position")
)

Fora

Use o método out() para criar um estágio de pipeline $out que grava todos os documentos na coleção especificada no mesmo banco de dados.

O exemplo a seguir grava os resultados do pipeline na coleção classic_movies :

Aggregates.out("classic_movies")

Merge

Use o método merge() para criar um estágio de pipeline $merge que funde todos os documentos na coleção especificada.

O exemplo a seguir mescla o pipeline na coleção nineties_movies utilizando as opções padrão:

Aggregates.merge("nineties_movies")

O exemplo a seguir mescla o pipeline na collection movie_ratings no reconhecimento de data center aggregation usando algumas opções não padrão que especificam para substituir o documento se year e title corresponderem, caso contrário, insira o documento:

Aggregates.merge(
    MongoNamespace("aggregation", "movie_ratings"),
    MergeOptions().uniqueIdentifier(listOf("year", "title"))
            .whenMatched(MergeOptions.WhenMatched.REPLACE)
        .whenNotMatched(MergeOptions.WhenNotMatched.INSERT)
)

GraphLookup

Use the graphLookup() method to create a $graphLookup pipeline stage that performs a recursive search on a specified collection to match a specified field in one document to a specified field of another document.

O exemplo a seguir utiliza a coleção contacts . Os dados são modelados usando a seguinte classe de dados Kotlin:

data class Users(
    val name: String,
    val friends: List<String>?,
    val hobbies: List<String>?
)

O exemplo calcula o gráfico de relatórios para usuários na collection contact , fazendo a correspondência recursiva do valor no campo friends com o campo name :

Aggregates.graphLookup(
    "contacts",
    "\$${Users::friends.name}", Users::friends.name, Users::name.name,
    "socialNetwork"
)

Usando GraphLookupOptions, você pode especificar a profundidade a ser recursada e o nome do campo de profundidade, se desejar. Neste exemplo, $graphLookup recursará até duas vezes e criará um campo chamado degrees com as informações de profundidade de recursão para cada documento.

Aggregates.graphLookup(
    "contacts",
    "\$${Users::friends.name}", Users::friends.name, Users::name.name,
    "socialNetwork",
    GraphLookupOptions().maxDepth(2).depthField("degrees")
)

Usando GraphLookupOptions, você pode especificar um filtro ao qual os documentos devem corresponder para que o MongoDB os inclua em sua pesquisa. Neste exemplo, apenas links com "golfe" no campo hobbies serão incluídos:

Aggregates.graphLookup(
    "contacts",
    "\$${Users::friends.name}", Users::friends.name, Users::name.name, "socialNetwork",
    GraphLookupOptions().maxDepth(1).restrictSearchWithMatch(
        Filters.eq(Users::hobbies.name, "golf")
    )
)

SortByCount

Use o método sortByCount() para criar um estágio de pipeline $sortByCount que agrupa documentos por uma determinada expressão e, em seguida, classifica esses grupos por contagem em ordem decrescente.

[
    { "$group": { "_id": <expression to group on>, "count": { "$sum": 1 } } },
    { "$sort": { "count": -1 } }
]

O exemplo a seguir agrupa documento na collection movies pelo campo genres e calcula a contagem para cada valor distinto:

Aggregates.sortByCount("\$${Movie::genres.name}"),

ReplaceRoot

Use o método replaceRoot() para criar um estágio de pipeline $replaceRoot que substitui cada documento de entrada pelo documento especificado.

O exemplo a seguir utiliza uma collection books fictício que contém dados modelados utilizando a seguinte classe de dados Kotlin:

data class Libro(val titulo: String)
data class Book(val title: String, val spanishTranslation: Libro)

Cada documento de entrada é substituído pelo documento aninhado no campo spanishTranslation :

Aggregates.replaceRoot("\$${Book::spanishTranslation.name}")

AddFields

Use the addFields() method to create an $addFields pipeline stage that adds new fields to documents.

O exemplo a seguir adiciona dois novos campos, watched e type, aos documentos de entrada na collection movie :

Aggregates.addFields(
    Field("watched", false),
    Field("type", "movie")
)

Contar

Use the count() method to create a $count pipeline stage that counts the number of documents that enter the stage, and assigns that value to a specified field name. If you do not specify a field, count() defaults the field name to "count".

{ "$group":{ "_id": 0, "count": { "$sum" : 1 } } }

O exemplo a seguir cria um estágio de pipeline que gera a contagem de documentos recebidos em um campo chamado "total":

Aggregates.count("total")

Balde

Use o método bucket() para criar um estágio de pipeline $bucket que automatize o agrupamento de dados em torno de valores de limite predefinidos.

Os exemplos a seguir usam dados modelados com a seguinte classe de dados Kotlin:

data class Screen(
    val id: String,
    val screenSize: Int,
    val manufacturer: String,
    val price: Double
)

Este exemplo cria um estágio de pipeline que agrupa os documentos recebidos com base no valor do campo screenSize , incluindo o limite inferior e excluindo o limite superior:

Aggregates.bucket("\$${Screen::screenSize.name}", listOf(0, 24, 32, 50, 70, 1000))

Utilize a classe BucketOptions para especificar um contêiner padrão para valores fora dos limites especificados e para especificar acumuladores adicionais.

O exemplo abaixo cria um estágio de pipeline que agrupa documentos de entrada com base no valor de seu campo screenSize, contando o número de documentos que entram em cada bucket, empurrando o valor de screenSize para um campo chamado matches, e capturando qualquer tamanho de tela maior que "70" em um bucket chamado "monster" para tamanhos de tela monstruosamente grandes:

Aggregates.bucket("\$${Screen::screenSize.name}", listOf(0, 24, 32, 50, 70),
    BucketOptions()
        .defaultBucket("monster")
        .output(
            Accumulators.sum("count", 1),
            Accumulators.push("matches", "\$${Screen::screenSize.name}")
        )
)

BucketAuto

Use the bucketAuto() method to create a $bucketAuto pipeline stage that automatically determines the boundaries of each bucket in its attempt to distribute the documents evenly into a specified number of buckets.

Os exemplos a seguir usam dados modelados com a seguinte classe de dados Kotlin:

data class Screen(
    val id: String,
    val screenSize: Int,
    val manufacturer: String,
    val price: Double
)

Este exemplo cria um estágio de pipeline que tentará criar e distribuir uniformemente documentos em 5 buckets usando o valor de seu campo price :

Aggregates.bucketAuto("\$${Screen::screenSize.name}", 5)

Utilize a classe BucketAutoOptions para especificar um esquema baseado em número preferido para definir valores de limite e especificar acumuladores adicionais.

O exemplo a seguir cria um estágio de pipeline que tentará criar e distribuir uniformemente documentos em 5 blocos usando o valor de seu campo price , definindo os limites de bloco em potências de 2 (2, 4, 8, 16, ...) . Ele também conta o número de documentos em cada bucket e calcula sua price média em um novo campo chamado avgPrice :

Aggregates.bucketAuto(
    "\$${Screen::price.name}", 5,
    BucketAutoOptions()
        .granularity(BucketGranularity.POWERSOF2)
        .output(Accumulators.sum("count", 1), Accumulators.avg("avgPrice", "\$${Screen::price.name}"))
        )

Facet

Use the facet() method to create a $facet pipeline stage that allows for the definition of parallel pipelines.

Os exemplos a seguir usam dados modelados com a seguinte classe de dados Kotlin:

data class Screen(
    val id: String,
    val screenSize: Int,
    val manufacturer: String,
    val price: Double
)

Este exemplo cria um estágio de pipeline que executa duas agregações paralelas:

• A primeira agregação distribui documentos recebidos em 5 grupos de acordo com seu campo screenSize.

• A segunda aggregation conta todos os fabricantes e gera sua contagem, limitada aos 5 principais.

Aggregates.facet(
    Facet(
        "Screen Sizes",
        Aggregates.bucketAuto(
            "\$${Screen::screenSize.name}",
            5,
            BucketAutoOptions().output(Accumulators.sum("count", 1))
        )
    ),
    Facet(
        "Manufacturer",
        Aggregates.sortByCount("\$${Screen::manufacturer.name}"),
        Aggregates.limit(5)
    )
)

DefinirWindowFields

Use o método setWindowFields() para criar um estágio de pipeline $setWindowFields que permite usar operadores de janela para realizar operações em uma extensão específica de documentos em uma coleção.

O exemplo a seguir utiliza uma collection weather fictício utilizando dados modelados com a seguinte classe de dados Kotlin:

data class Weather(
    val localityId: String,
    val measurementDateTime: LocalDateTime,
    val rainfall: Double,
    val temperature: Double
)

O exemplo cria um estágio de tubulação que calcula a precipitação acumulada e a temperatura média no mês passado para cada localidade a partir de medições mais refinadas apresentadas nos campos rainfall e temperature :

val pastMonth = Windows.timeRange(-1, MongoTimeUnit.MONTH, Windows.Bound.CURRENT)
val resultsFlow = weatherCollection.aggregate<Document>(
    listOf(
       Aggregates.setWindowFields("\$${Weather::localityId.name}",
           Sorts.ascending(Weather::measurementDateTime.name),
           WindowOutputFields.sum(
               "monthlyRainfall",
               "\$${Weather::rainfall.name}",
               pastMonth
           ),
           WindowOutputFields.avg(
               "monthlyAvgTemp",
               "\$${Weather::temperature.name}",
               pastMonth
           )
       )
    )

Densificar

Utilize o método densify() para criar um estágio de pipeline $densify que gera uma sequência de documentos para abranger um intervalo especificado.

Considere os seguintes documentos recuperados do conjunto de dados meteorológicos de amostra do Atlas que contêm medições para um campo de position semelhante, espaçados em uma hora:

Document{{ _id=5553a..., position=Document{{type=Point, coordinates=[-47.9, 47.6]}}, ts=Mon Mar 05 08:00:00 EST 1984, ... }}
Document{{ _id=5553b..., position=Document{{type=Point, coordinates=[-47.9, 47.6]}}, ts=Mon Mar 05 09:00:00 EST 1984, ... }}

Esses documentos são modelados usando a seguinte classe de dados Kotlin:

data class Weather(
    @BsonId val id: ObjectId = ObjectId(),
    val position: Point,
    val ts: LocalDateTime
)

Suponha que você precise criar um estágio de pipeline que execute as seguintes ações nesses documentos:

• Adicione um documento a cada intervalo de ts minutos para o qual um valor de ainda não existe.

• Agrupe os documentos pelo campo position.

A chamada para o construtor de estágio de agregação do densify() que realiza essas ações deve se assemelhar ao seguinte:

Aggregates.densify(
    "ts",
    DensifyRange.partitionRangeWithStep(15, MongoTimeUnit.MINUTE),
    DensifyOptions.densifyOptions().partitionByFields("Position.coordinates")
)

A saída a seguir destaca os documentos gerados pelo estágio agregado que contêm valores ts a cada 15 minutos entre os documentos existentes:

Document{{ _id=5553a..., position=Document{{type=Point, coordinates=[-47.9, 47.6]}}, ts=Mon Mar 05 08:00:00 EST 1984, ... }}
Document{{ position=Document{{coordinates=[-47.9, 47.6]}}, ts=Mon Mar 05 08:15:00 EST 1984 }}
Document{{ position=Document{{coordinates=[-47.9, 47.6]}}, ts=Mon Mar 05 08:30:00 EST 1984 }}
Document{{ position=Document{{coordinates=[-47.9, 47.6]}}, ts=Mon Mar 05 08:45:00 EST 1984 }}
Document{{ _id=5553b..., position=Document{{type=Point, coordinates=[-47.9, 47.6]}}, ts=Mon Mar 05 09:00:00 EST 1984, ... }}

Consulte a documentação da API do pacote densify para obter mais informações.

Fill

Use the fill() method to create a $fill pipeline stage that populates null and missing field values.

Considere os seguintes documentos que contêm medições de temperatura e pressão de ar em um intervalo de hora em hora:

Document{{_id=6308a..., hour=1, temperature=23C, air_pressure=29.74}}
Document{{_id=6308b..., hour=2, temperature=23.5C}}
Document{{_id=6308c..., hour=3, temperature=null, air_pressure=29.76}}

Esses documentos são modelados usando a seguinte classe de dados Kotlin:

data class Weather(
    @BsonId val id: ObjectId = ObjectId(),
    val hour: Int,
    val temperature: String?,
    val air_pressure: Double?
)

Suponha que você precisasse preencher os pontos de dados ausentes de temperatura e pressão atmosférica nos documentos da seguinte forma:

• Preencha o campo air_pressure para hora "2" utilizando interpolação linear para calcular o valor.

• Defina o valor temperature ausente para " 23.6C " por hora "3".

A chamada para o construtor de estágio de agregação do fill() que realiza estas ações se assemelha ao seguinte:

val resultsFlow = weatherCollection.aggregate<Weather>(
    listOf(
        Aggregates.fill(
            FillOptions.fillOptions().sortBy(Sorts.ascending(Weather::hour.name)),
            FillOutputField.value(Weather::temperature.name, "23.6C"),
            FillOutputField.linear(Weather::air_pressure.name)
        )
    )
)
resultsFlow.collect { println(it) }

Weather(id=6308a..., hour=1, temperature=23C, air_pressure=29.74)
Weather(id=6308b..., hour=2, temperature=23.5C, air_pressure=29.75)
Weather(id=6308b..., hour=3, temperature=23.6C, air_pressure=29.76)

Consulte a documentação completa da API do pacote para obter mais informações.

Pesquisa de texto completo do Atlas

Use the search() method to create a $search pipeline stage that specifies a full-text search of one or more fields.

O exemplo abaixo cria um estágio de pipeline que pesquisa no campo title na collection movies o texto que contém a palavra "Future":

Aggregates.search(
    SearchOperator.text(
        SearchPath.fieldPath(Movie::title.name), "Future"
    ),
    SearchOptions.searchOptions().index("title")
)

Saiba mais sobre os construtores na documentação da API do pacote de pesquisa.

{
  "mappings": {
    "dynamic": true,
    "fields": {
      "title": {
        "analyzer": "lucene.keyword",
        "type": "string"
      },
      "genres": {
        "normalizer": "lowercase",
        "type": "token"
      }
    }
  }
}

data class Results(val title: String, val year: Int, val genres: List<String>)
val searchStage = Aggregates.search(
    SearchOperator.compound()
        .filter(
            listOf(
                SearchOperator.`in`(SearchPath.fieldPath(Movie::genres.name), listOf("Comedy")),
                SearchOperator.phrase(SearchPath.fieldPath(Movie::fullplot.name), "new york"),
                SearchOperator.numberRange(SearchPath.fieldPath(Movie::year.name)).gtLt(1950, 2000),
                SearchOperator.wildcard(SearchPath.fieldPath(Movie::title.name), "Love *")
            )
        )
)
val projectStage = Aggregates.project(
    Projections.include(Movie::title.name, Movie::year.name, Movie::genres.name))
val pipeline = listOf(searchStage, projectStage)
val resultsFlow = ftsCollection.aggregate<Results>(pipeline)
resultsFlow.collect { println(it) }

Results(title=Love at First Bite, year=1979, genres=[Comedy, Romance])
Results(title=Love Affair, year=1994, genres=[Comedy, Drama])

Metadados de pesquisa do MongoDB

Use the searchMeta() method to create a $searchMeta pipeline stage which returns only the metadata part of the results from Atlas full-text search queries.

O exemplo a seguir mostra os metadados count para um estágio de agregação do MongoDB pesquisa:

Aggregates.searchMeta(
    SearchOperator.near(1985, 2, SearchPath.fieldPath(Movie::year.name)),
    SearchOptions.searchOptions().index("year")
)

Saiba mais sobre esse assistente na documentação da API searchMeta().