集計ビルダ

Overview

このガイドでは、 Java Reactive Streams ドライバーで集計パイプラインライン ステージを構築する静的ファクトリー メソッドを提供する Aggregatesクラスの使用方法を学習できます。

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

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

• Aggregates

• Filters

• Projections

• Sorts

• Accumulators

次のコードは、前述のクラスのメソッドをインポートする方法を示しています。

import static java.util.Arrays.asList;
import static com.mongodb.client.model.Accumulators.*;
import static com.mongodb.client.model.Aggregates.*;
import static com.mongodb.client.model.Filters.*;
import static com.mongodb.client.model.Projections.*;
import static com.mongodb.client.model.Sorts.*;
import static com.mongodb.client.model.search.SearchPath.fieldPath;

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

Bson matchStage = match(eq("some_field", "some_criteria"));
Bson sortByCountStage = sortByCount("some_field");
Publisher<Document> results = collection.aggregate(asList(matchStage, sortByCountStage));
Flux.from(results).collectList().block();

一致

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

次の例では、 titleフィールドが「シャードのシャーディング」に等しいすべてのドキュメントに一致するパイプライン ステージを作成します。

match(eq("title", "The Shawshank Redemption"));            

プロジェクト

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

次の例では、 _idフィールドを除外するものの、 titleフィールドとplotフィールドを含むパイプライン ステージを作成します。

project(fields(include("title", "plot"), excludeId()));

project(fields(computed("rating", "$rated"), excludeId()));

ドキュメント

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

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

documents(asList(
        new Document("title", "The Shawshank Redemption"),
        new Document("title", "Back to the Future"),
        new Document("title", "Jurassic Park")));

サンプル

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

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

sample(5);

Sort

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

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

sort(orderBy(descending("year"), ascending("title")));

スキップ

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

次の例では、最初の5ドキュメントをスキップするパイプライン ステージを作成しています。

skip(5);

Limit

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

次の例では、ドキュメント数を10に制限するパイプライン ステージを作成しています。

limit(10);

ルックアップ

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

lookup("comments", "_id", "movie_id", "joined_comments");

List<Variable<Object>> variables = asList(
    new Variable<>("genre", new Document("$arrayElemAt", asList("$genres", 0))),
    new Variable<>("rating", "$imdb.rating")
);
List<Bson> pipeline = asList(
    match(expr(new Document("$and",
        asList(
            new Document("$in", asList("$$genre", "$genres")),
            new Document("$gt", asList("$imdb.rating", "$$rating"))
        )))),
    project(fields(include("title", "imdb.rating"), excludeId())));
Bson similarHigherRatedLookup = lookup("movies", variables, pipeline, "similar_higher_rated");

グループ

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

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

group("$customerId", sum("totalQuantity", "$quantity"), avg("averageQuantity", "$quantity"));

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

Ticket-N アキュムレータ

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

• minN()
• maxN()
• firstN()
• lastN()
• top()
• topN()
• bottom()
• bottomN()

でアキュムレータ集計パイプラインステージについては、サーバー マニュアルの「 アキュムレータ 」のページから参照してください。

group(
    "$year",
    minN(
        "lowest_three_ratings",
        new BsonString("$imdb.rating"),
        3
        ));

group(
    "$year",
    maxN(
        "highest_two_ratings",
        new BsonString("$imdb.rating"),
        2
        ));

group(
    "$year",
    firstN(
        "first_four_movies",
        new BsonString("$title"),
        4
        ));

group(
    "$year",
    lastN(
        "last_three_movies",
        new BsonString("$title"),
        3
        ));

group(
    "$year", 
    top(
        "top_rated_movie",
        descending("imdb.rating"),
        asList(new BsonString("$title"), new BsonString("$imdb.rating"))
        ));

group(
    "$year", 
    topN(
        "longest_three_movies",
        descending("runtime"),
        asList(new BsonString("$title"), new BsonString("$runtime")),
        3
        ));

group(
    "$year", 
    bottom(
        "shortest_movies",
        descending("runtime"),
        asList(new BsonString("$title"), new BsonString("$runtime"))
        ));

group(
    "$year", 
    bottomN(
        "lowest_rated_two_movies",
        descending("imdb.rating"),
        asList(new BsonString("$title"), new BsonString("$imdb.rating")),
        2
        ));

Unwind

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

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

unwind("$sizes");

次の例では、配列フィールドの欠落値または null 値を持つドキュメント、または配列が空のドキュメントを保持します。

unwind("$sizes", new UnwindOptions().preserveNullAndEmptyArrays(true));

次の例には、"position"というフィールドに配列インデックスが含まれています。

unwind("$sizes", new UnwindOptions().includeArrayIndex("position"));

アウト

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

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

out("authors");

merge

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

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

merge("authors");

次の例では、 dateとcustomerIdの両方が一致する場合はドキュメントを置き換え、それ以外の場合はドキュメントを挿入するオプションを使用して、パイプラインをreportingデータベースのcustomersコレクションにマージします。

merge(new MongoNamespace("reporting", "customers"),
new MergeOptions().uniqueIdentifier(asList("date", "customerId"))
        .whenMatched(MergeOptions.WhenMatched.REPLACE)
        .whenNotMatched(MergeOptions.WhenNotMatched.INSERT));

GraphLookup

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

次の例では、 contactsコレクション内のユーザーのソーシャル ネットワーク グラフを計算し、 friendsフィールドの値をnameフィールドに再帰的に照合します。

graphLookup("contacts", "$friends", "friends", "name", "socialNetwork");

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

graphLookup("contacts", "$friends", "friends", "name", "socialNetwork",
new GraphLookupOptions().maxDepth(2).depthField("degrees"));

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

graphLookup("contacts", "$friends", "friends", "name", "socialNetwork",
new GraphLookupOptions().maxDepth(1).restrictSearchWithMatch(eq("hobbies", "golf")));

カウントによる並べ替え

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

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

次の例では、ドキュメントをフィールドxの切り捨てられた値でグループ化し、個別の値ごとにカウントを計算します。

sortByCount(new Document("$floor", "$x"));

ReplaceRoot

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

次の例では、各入力ドキュメントをspanish_translationフィールドにあるネストされたドキュメントに置き換えます。

replaceRoot("$spanish_translation");

AddFields

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

次の例では、入力ドキュメントにaとbの 2 つの新しいフィールドを追加します。

addFields(new Field("a", 1), new Field("b", 2));

数

count()メソッドを使用して、ステージに入るドキュメントの数をカウントし、その値を指定されたフィールド名に割り当てる $countパイプラインステージを作成します。フィールド名を指定しない場合、count() はデフォルトでフィールド名を「count」に設定します。

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

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

count("total");

バケット

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

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

bucket("$screenSize", asList(0, 24, 32, 50, 70, 200));

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

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

bucket("$screenSize", asList(0, 24, 32, 50, 70),
    new BucketOptions().defaultBucket("monster").output(sum("count", 1), push("matches", "$screenSize")));

BucketAuto

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

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

bucketAuto("$price", 10);

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

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

bucketAuto("$price", 10, new BucketAutoOptions().granularity(BucketGranularity.POWERSOF2)
.output(sum("count", 1), avg("avgPrice", "$price")));

Facet

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

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

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

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

facet(new Facet("Screen Sizes",
    bucketAuto("$attributes.screen_size", 5, new BucketAutoOptions().output(sum("count", 1)))),
    new Facet("Manufacturer", sortByCount("$attributes.manufacturer"), limit(5)));

SetWindowFields

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

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

Window pastMonth = Windows.timeRange(-1, MongoTimeUnit.MONTH, Windows.Bound.CURRENT);
setWindowFields("$localityId", ascending("measurementDateTime"),
    WindowOutputFields.sum("monthlyRainfall", "$rainfall", pastMonth),
    WindowOutputFields.avg("monthlyAvgTemp", "$temperature", 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, ... }}

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

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

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

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

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

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

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

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

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

fill(
    FillOptions.fillOptions().sortBy(ascending("hour")),
    FillOutputField.value("temperature", "23.6C"),
    FillOutputField.linear("air_pressure")
);

次の出力では、 集計ステージによって入力されるフィールドを含むドキュメントが強調表示されています。

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

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

MongoDB Search

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

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

Bson textSearch = search(
    SearchOperator.text(
        fieldPath("title"), "Future"));

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

MongoDB 検索する メタデータ

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

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

searchMeta(
    SearchOperator.near(2010, 1, fieldPath("year")));