集計ビルダ

Overview

このガイドでは、 MongoDB Kotlinドライバーで集計パイプラインライン ステージを構築する静的ファクトリー メソッドを提供する Aggregatesクラスの使用方法を説明します。

集計の詳細については、 集計ガイド をご覧ください。

このページの例では、次のクラスのメソッドのインポートを前提としています。

• 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

これらのメソッドを使用してパイプライン ステージを構築し、それらをリストとして集計で指定します。

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

このガイドの多くのAggregation例では、 Atlas sample_mflix.movies データセットを使用します。 このコレクション内のドキュメントは、 Kotlin ドライバーで使用する次のMovieデータ クラスによってモデル化されます。

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

一致

match()メソッドを使用して、指定されたクエリフィルターと受信ドキュメントを照合する$matchパイプライン ステージを作成し、一致しないドキュメントをフィルタリングで除外します。

moviesのtitle

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

プロジェクト

project()メソッドを使用して、指定されたドキュメント フィールドをプロジェクションする$projectパイプライン ステージを作成します。 集計でのフィールド プロジェクションは、クエリの フィールド プロジェクションと同じルールに従います。

次の例では、 titleフィールドとplotフィールドを含み、 _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()
    )
)

ドキュメント

documents()メソッドを使用して、入力値からリテラル ドキュメントを返す$documentsパイプライン ステージを作成します。

次の例では、 titleフィールドを持つmoviesコレクションにサンプル ドキュメントを作成するパイプライン ステージを作成します。

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

サンプル

sample()メソッドを使用して、入力からドキュメントをランダムに選択するための$sampleパイプライン ステージを作成します。

次の例では、 moviesコレクションからランダムに 5 つのドキュメントを選択するパイプライン ステージを作成します。

Aggregates.sample(5)

Sort

sort()メソッドを使用して、指定された条件で並べ替えるための$sortパイプライン ステージを作成します。

次の例では、 yearフィールドの値に従って降順でソートし、次にtitleフィールドの値の昇順でソートするパイプライン ステージを作成します。

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

スキップ

skip()メソッドを使用して$skipパイプライン ステージを作成し、ドキュメントを次のステージに渡す前に指定された数のドキュメントをスキップします。

次の例では、 moviesコレクション内の最初の5ドキュメントをスキップするパイプライン ステージを作成します。

Aggregates.skip(5)

Limit

次のステージに渡されるドキュメントの数を制限するには、 $limitパイプライン ステージを使用します。

次の例では、 moviesコレクションから返されるドキュメントの数を4に制限するパイプライン ステージを作成しています。

Aggregates.limit(4)

ルックアップ

lookup()メソッドを使用して$lookupパイプライン ステージを作成し、2 つのコレクション間で結合と非相関サブクエリを実行します。

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

グループ

group()メソッドを使用して$groupパイプライン ステージを作成し、指定された式でドキュメントをグループ化し、個別のグループごとにドキュメントを出力します。

次の例では、 ordersコレクション内のドキュメントをcustomerIdフィールドの値でグループ化するパイプライン ステージを作成します。 各グループは、 orderedフィールドの値の合計と平均をtotalQuantityフィールドとaverageQuantityフィールドに累積します。

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

アキュムレータ演算子の詳細については、サーバー マニュアルの「アキュムレータ 」セクションを参照してください。

Ticket-N アキュムレータ

選択可能アキュムレータ は、特定の順序指定された上位要素と最下位要素を返す集計アキュムレーション演算子です。 次のいずれかのビルダを使用して、集計アキュムレーション演算子を作成します。

• minN()

• maxN()

• firstN()

• lastN()

• top()

• topN()

• bottom()

• bottomN()

アキュムレータ演算子を使用できる集計パイプライン ステージについては、サーバー マニュアルの「アキュムレータ 」セクションを参照してください。

選択可能アキュムレータの例では、 sample-mflixデータベース内のmoviesコレクションのドキュメントを使用します。

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

unwind()メソッドを使用して$unwindパイプライン ステージを作成し、入力ドキュメントから配列フィールドを分解し、配列要素ごとに出力ドキュメントを作成します。

次の例では、 lowestRatedTwoMovies配列内の各要素のドキュメントを作成します。

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

配列フィールドの欠落値またはnull値、または配列が空のドキュメントを保持するには:

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

配列インデックスを含めるには（この例では"position"というフィールド内）

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

アウト

out()メソッドを使用して、すべてのドキュメントを同じデータベース内の指定されたコレクションに書込む$outパイプライン ステージを作成します。

次の例では、パイプラインの結果をclassic_moviesコレクションに書き込みます。

Aggregates.out("classic_movies")

merge

merge()メソッドを使用して、すべてのドキュメントを指定されたコレクションにマージする$mergeパイプライン ステージを作成します。

次の例では、デフォルトの オプションを使用してパイプラインをnineties_moviesコレクションにマージします。

Aggregates.merge("nineties_movies")

次の例では、 yearとtitleの両方が一致する場合はドキュメントを置き換えるために指定するデフォルト以外のオプションを使用してパイプラインをaggregationデータベースのmovie_ratingsコレクションにマージします。それ以外の場合はドキュメントを挿入します。

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

GraphLookup

graphLookup()メソッドを使用して、指定されたコレクションに対して再帰検索を実行し、1 つのドキュメント内の指定されたフィールドを別のドキュメントの指定されたフィールドと照合する$graphLookupパイプライン ステージを作成します。

次の例では、 contactsコレクションを使用します。 データは、次の Kotlin データ クラスを使用してモデル化されます。

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

この例では、 contactコレクション内のユーザーのレポート作成グラフを計算し、 friendsフィールドの値をnameフィールドに再帰的に照合します。

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

GraphLookupOptions を使用すると、必要に応じて再帰する深度と深度フィールドの名前を指定できます。 この例では 、$graphLookup は最大 2 回再帰し、すべてのドキュメントの再帰深度情報を持つ degrees というフィールドを作成します。

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

GraphLookupOptionsを使用すると、MongoDB が検索にドキュメントを含めるために一致する必要があるフィルターを指定できます。 この例では、 hobbiesフィールドに「golf」が含まれるリンクのみが含まれます。

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

カウントによる並べ替え

sortByCount()メソッドを使用して、特定の式でドキュメントをグループ化し、これらのグループをカウントで降順にソートする$sortByCountパイプライン ステージを作成します。

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

次の例では、 moviesコレクション内のドキュメントをgenresフィールドでグループ化し、個別の 値ごとにカウントを計算します。

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

ReplaceRoot

replaceRoot()メソッドを使用して、各入力ドキュメントを指定されたドキュメントで置き換える$replaceRootパイプライン ステージを作成します。

次の例では、次の Kotlin データ クラスを使用してモデル化されたデータを含む架空のbooksコレクションを使用します。

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

各入力ドキュメントは、 spanishTranslationフィールドのネストされたドキュメントに置き換えられます。

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

AddFields

addFields()メソッドを使用して、ドキュメントに新しいフィールドを追加する$addFieldsパイプライン ステージを作成します。

次の例では、 movieコレクションの入力ドキュメントに 2 つの新しいフィールド （ watchedとtype ）を追加します。

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

数

count()メソッドを使用して、ステージに入るドキュメントの数をカウントし、その値を指定されたフィールド名に割り当てる$countパイプライン ステージを作成します。 If you do not specify a field, count() defaults the field name to "count".

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

次の例では、「total」というフィールドに受信したドキュメントの数を出力するパイプライン ステージを作成します。

Aggregates.count("total")

バケット

bucket()メソッドを使用して、事前定義された境界値の周囲のデータのバケットを自動化する$bucketパイプライン ステージを作成します。

次の例では、次の Kotlin データ クラスでモデル化されたデータを使用します。

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

この例では、受信ドキュメントをscreenSizeフィールドの値に基づいてグループ化するパイプライン ステージを作成します。このステージは下限を含み、上限を含まないものです。

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

指定された境界外の値のデフォルト バケットを指定し、追加のアキュムレータを指定するには、 BucketOptionsクラスを使用します。

次の例では、 screenSizeフィールドの値に基づいて受信ドキュメントをグループ化し、各バケットに含まれるドキュメントの数をカウントし、 screenSizeの値をmatchesというフィールドにプッシュして、次を取得するパイプライン ステージを作成します: 「70」を超える任意の画面サイズを、平均して大きな画面サイズ用の「mongostat」と呼ばれるバケットに格納します。

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

bucketAuto()メソッドを使用して$bucketAutoパイプライン ステージを作成します。このステージは、指定された数のバケットにドキュメントを均等に分散するために各バケットの境界を自動的に決定します。

次の例では、次の Kotlin データ クラスでモデル化されたデータを使用します。

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

この例では、 priceフィールドの値を使用して、ドキュメントを作成し、5 つのバケットに均等に分散するパイプライン ステージを作成します。

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

BucketAutoOptionsクラスを使用して、境界値を設定するための推奨数値ベースのスキームを指定し、追加のアキュムレータを指定します。

次の例では、 priceフィールドの値を使用して 5 つのバケットにドキュメントを作成し、均等に分散するパイプライン ステージを作成し、バケット境界を 2 の累乗（2、4、8、16、...）に設定します。 。 また、各バケット内のドキュメント数をカウントし、 avgPriceという新しいフィールドでそれらの平均priceを計算します。

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

Facet

facet()メソッドを使用して$facetパイプライン ステージを作成し、並列パイプラインの定義を可能にします。

次の例では、次の Kotlin データ クラスでモデル化されたデータを使用します。

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

この例では、2 つの並列集計を実行するパイプライン ステージを作成しています。

• 最初の集計では、受信したドキュメントをscreenSizeフィールドに従って 5 つのグループに分散します。

• 2 番目の集計では、すべてのメーカーをカウントし、上位 5 に限定したその数を返します。

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

SetWindowFields

setWindowFields()メソッドを使用して$setWindowFieldsパイプライン ステージを作成し、ウィンドウ演算子がコレクション内の指定された範囲のドキュメントに対して操作を実行できるようにします。

次の例では、次の Kotlin データ クラスでモデル化されたデータを使用する架空のweatherコレクションを使用します。

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

この例では、 フィールドと フィールドに表示されるより詳細な測定値から、各地域の過去 1 か月間の累積降量と平均温度を計算するパイプライン ステージを作成します。rainfalltemperature

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

密度

densify()メソッドを使用して、指定された間隔にわたるドキュメントのシーケンスを生成する$densifyパイプライン ステージを作成します。

Atlas サンプル 気象 データセットから取得され、1 時間間隔で配置された同様のpositionフィールドの測定値を含む次のドキュメントを検討してください。

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, ... }}

これらのドキュメントは、次の Kotlin データ クラスを使用してモデル化されます。

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

これらのドキュメントに対して次のアクションを実行するパイプライン ステージを作成する必要があるとします。

• ts値がまだ存在しないドキュメントを 15 分ごとに追加します。

• ドキュメントをpositionフィールドでグループ化します。

これらのアクションを実行するdensify()集計ステージ ビルダへの呼び出しは、次のようになります。

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

次の出力では、既存のドキュメント間で 15 分ごとのts値を含む、 集計ステージによって生成されたドキュメントが強調表示されています。

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, ... }}

詳細については、densifyパッケージAPIドキュメントを参照してください。

Fill

fill()メソッドを使用して、 nullと欠落しているフィールド値を入力する$fillパイプライン ステージを作成します。

1 時間ごとの温度と温度の測定値を含む次のドキュメントを検討してください。

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

これらのドキュメントは、次の Kotlin データ クラスを使用してモデル化されます。

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

次のように、ドキュメントに欠落している温度と温度のデータ ポイントを入力する必要があるとします。

• 線形補間 を使用して値を計算し、 air_pressureフィールドに時間 "2" を入力します。

• 欠落しているtemperature値を「23.6C」に設定します 時間の「3」。

これらのアクションを実行するfill()集計ステージ ビルダへの呼び出しは次のようになります。

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)

詳細については、fillパッケージAPIドキュメントを参照してください。

Atlas 全文検索

search()メソッドを使用して、1 つ以上のフィールドの全文検索を指定する$searchパイプライン ステージを作成します。

次の例では、 moviesコレクションのtitleフィールドで「feature」という単語を含むテキストを検索するパイプライン ステージを作成します。

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

ビルダの詳細については、検索パッケージAPIドキュメントを参照してください。

Atlas Search メタデータ

searchMeta()メソッドを使用して、Atlas 全文検索クエリの結果のメタデータ部分のみを返す$searchMetaパイプライン ステージを作成します。

次の例では、Atlas Search 集計ステージのcountメタデータを示しています。

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

このヘルパーの詳細については、searchMeta() APIドキュメント を参照してください。