Construtores de agregados

Visão geral

Neste guia, você pode aprender como usar a classeAgregados que fornece métodos de fábrica estáticos que constroem estágios de pipeline de agregação no driver Kotlin MongoDB .

Para uma introdução mais completa à Agregação, consulte nossoGuia de Agregaçãodo .

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()

Muitos exemplos do Aggregation neste guia utilizam o conjunto de dados sample_mflix.movies do Atlas. Os documentos nesta coleção são modelados pela seguinte classe de dados Movie para uso com o driver Kotlin:

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
){
    data class IMDB(
        val rating: Double
    )
}

corresponder

Use o método match() para criar um estágio de pipeline $match que faça a correspondência dos documentos recebidos com o filtro de queries especificado, filtrando os documentos que não correspondem.

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 o método documents() para criar um estágio de pipeline $documents que retorna documentos literais a partir de valores de entrada.

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 o método skip() para criar um estágio de pipeline $skip para ignorar o número especificado de documentos antes de passar os documentos para o próximo estágio.

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 o método lookup() para criar um estágio de pipeline $lookup para realizar fusões e subconsultas não correlacionadas entre duas coleções.

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 o método group() para criar um estágio de pipeline $group para agrupar documentos com base em uma determinada expressão e gerar um documento para cada agrupamento distinto.

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}")
)

Saiba mais sobre os operadores acumuladores na seção Manual do servidor em Acumuladores.

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 pipeline de agregação 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 o método unwind() para criar um estágio de pipeline $unwind para desconstruir um campo de array a partir de documentos de entrada, criando um documento de saída para cada elemento de array.

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 o método graphLookup() para criar um estágio de pipeline $graphLookup que executa uma pesquisa recursiva em uma coleção especificada para corresponder a um campo especificado em um documento a um campo especificado de outro documento.

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 o método addFields() para criar um estágio de pipeline $addFields que adiciona novos campos a documentos.

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 o método count() para criar um estágio de pipeline $count que conta o número de documentos que entram no estágio e atribui esse valor a um nome de campo especificado. Se você não especificar um campo, count() padronizará o nome do campo para "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 o método bucketAuto() para criar um estágio de pipeline $bucketAuto que determina automaticamente os limites de cada bucket em sua tentativa de distribuir os documentos uniformemente em um número especificado de 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 o método facet() para criar um estágio de pipeline $facet que permitirá a definição de pipelines paralelos.

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 blocos 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 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 o método fill() para criar um estágio de pipeline $fill que preencha null e os valores de campo ausentes.

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

Utilize o método search() para criar um estágio de pipeline $search que especifica uma Full Text Search de um ou mais campos.

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.

Metadados de Atlas Search

Use o método searchMeta() para criar um estágio de pipeline $searchMeta que retorna somente a parte de metadados dos resultados das queries de Full Text Search do Atlas.

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

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().