集計ビルダ

Overview

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

集計の詳細については、 集計ガイドをご覧ください。 集計で使用する、検出可能でタイプセーフな Java メソッドを使用して式操作を作成する方法については、「集計式操作 」を参照してください。

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.Accumulators.*;
import static com.mongodb.client.model.search.SearchPath.fieldPath;
import static com.mongodb.client.model.search.VectorSearchOptions.exactVectorSearchOptions;
import static java.util.Arrays.asList;

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

Bson matchStage = match(eq("some_field", "some_criteria"));
Bson sortByCountStage = sortByCount("some_field");
collection.aggregate(asList(matchStage, sortByCountStage)).forEach(doc -> System.out.println(doc));

一致

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

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

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

プロジェクト

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

次の例では、 _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パイプライン ステージを作成します。

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

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<String>> variables = asList(new Variable<>("order_item", "$item"),
        new Variable<>("order_qty", "$ordered"));
List<Bson> pipeline = asList(
        match(expr(new Document("$and",
                asList(new Document("$eq", asList("$$order_item", "$stock_item")),
                        new Document("$gte", asList("$instock", "$$order_qty")))))),
        project(fields(exclude("stock_item"), excludeId())));
Bson innerJoinLookup = lookup("warehouses", variables, pipeline, "stockdata");

グループ

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パイプライン ステージを作成します。 If you do not specify a field, count() defaults the field name to "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", Sorts.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}}

フィルター パッケージ API ドキュメント を参照してください 詳しくは、 を参照してください。

Atlas 全文検索

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

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

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

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

Atlas Search メタデータ

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

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

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

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

Atlas Vector Search

vectorSearch()メソッドを使用して、 セマンティック検索 を指定する $vectorSearch パイプライン ステージを作成します。セマンティック検索とは、意味が同様の情報を検索するタイプの検索です。

この機能を使用するには、ベクトル検索インデックスを設定し、ベクトル埋め込みにインデックスを付ける必要があります。 ベクトル検索インデックスをプログラムで作成する方法の詳細については、「 インデックス ガイド 」の「 Atlas Search と Vector Search インデックス 」セクションを参照してください。 ベクトル埋め込みの詳細については、「ベクトル検索のためのベクトル埋め込みのインデックスを作成する方法 」を参照してください。

次の例は、 メソッドと メソッドを使用してベクトル検索スコアを計算する集計パイプラインを構築する方法を示しています。vectorSearch()project()

List<Double> queryVector = (asList(-0.0072121937, -0.030757688, -0.012945653));
String indexName = "mflix_movies_embedding_index";
FieldSearchPath fieldSearchPath = fieldPath("plot_embedding");
int limit = 1;
VectorSearchOptions options = exactVectorSearchOptions().filter(gte("year", 2016));
List<Bson> pipeline = asList(
        vectorSearch(
                fieldSearchPath,
                queryVector,
                indexName,
                limit,
                options),
        project(
                metaVectorSearchScore("vectorSearchScore")));

次の例は、前述の集計パイプラインの結果からスコアを印刷する方法を示しています。

Document found = collection.aggregate(pipeline).first();
double score = found.getDouble("vectorSearchScore").doubleValue();
System.out.println("vectorSearch score: " + score);

vectorSearch() のこのヘルパーの詳細 API ドキュメント。