MongoDB Java API медленное считывание

Ответ 1

Вы не указали свой прецедент, поэтому очень сложно рассказать вам, как настроить ваш запрос. (Т.е.: Кто хотел бы загружать 800mil строк за раз только для подсчета?).

Учитывая вашу схему, я думаю, что ваши данные почти доступны только для чтения, и ваша задача связана с агрегацией данных.

Ваша текущая работа просто считывает данные (скорее всего, ваш драйвер будет читать в пакетном режиме), а затем остановится, затем выполните некоторые вычисления (черт возьми, инт-обертка используется для увеличения времени обработки больше), а затем повторите. Это не очень хороший подход. БД не магически быстро, если вы не получите доступ к нему правильным способом.

Если вычисление не слишком сложное, я предлагаю вам использовать структуру агрегации вместо загрузки всех в оперативную память.

Что-то просто вы должны рассмотреть, чтобы улучшить вашу агрегацию:

1. Разделите свой набор данных на меньший набор. (Например: раздел по date, раздел по exchange...). Добавьте индекс для поддержки этого раздела и выполните агрегирование на разделе, а затем объедините результат (типичный подход с делением-n-покорением)
2. Для проекта нужны только поля
3. Отфильтруйте ненужный документ (если это возможно)
4. Разрешить diskusage, если вы не можете выполнить агрегацию в памяти (если вы нажмете предел 100 Мбайт на каждый пинилин).
5. Используйте встроенный конвейер для ускорения вычислений (например: $count для вашего примера)

Если ваши вычисления слишком сложны, что вы не можете выразить с помощью структуры агрегации, используйте mapReduce. Он работает на процессе mongod и данные не нуждаются в передаче по сети в вашу память.

обновленный

Похоже, вы хотите сделать OLAP-обработку, и вы застряли на шаге ETL.

Вам не нужно и приходится каждый раз загружать все OLTP-данные в OLAP. Нужно только загружать новые изменения в хранилище данных. Тогда первая загрузка/сброс данных занимает больше времени, является нормальным и приемлемым.

Для первой загрузки вы должны учитывать следующие моменты:

1. Divide-N-Conquer, опять же, разбивает ваши данные на меньший набор данных (с предикатом, похожим на метку даты/обмена/запаса...)
2. Проведите параллельное вычисление, затем соедините результат (вам нужно правильно разбить ваш набор данных)
3. Выполняйте вычисления в пакетном режиме вместо обработки в forEach: затем загрузите раздел данных, а не вычисляйте один за другим.

Ответ 2

collection.find().forEach((Block<Document>) document → count.increment());

Эта строка может сбрасывать много времени с момента итерации более 250 тыс. Записей в памяти.

Чтобы быстро проверить, если это так, вы можете попробовать это -

long start1 = System.currentTimeMillis();
List<Document> documents = collection.find();
System.out.println(System.currentTimeMillis() - start1);

long start2 = System.currentTimeMillis();
documents.forEach((Block<Document>) document -> count.increment());
System.out.println(System.currentTimeMillis() - start2);

Это поможет вам понять, сколько времени требуется на получение документов из базы данных и сколько времени занимает итерация.

Ответ 3

То, что я думаю, должно было сделать в вашем случае, было простым решением, и одновременно эффективным способом является максимизация общей пропускной способности с помощью parallelCollectionScan

Позволяет приложениям использовать несколько параллельных курсоров при чтении всех документов из коллекции, тем самым увеличивая пропускную способность. Команда parallelCollectionScan возвращает документ, содержащий массив информации о курсоре.

Каждый курсор обеспечивает доступ к возврату частичного набора документов из коллекции. Итерация каждого курсора возвращает каждый документ в коллекции. Курсоры не содержат результатов команды базы данных. Результат команды базы данных идентифицирует курсоры, но не содержит или не составляет курсоров.

Простой пример с parallelCollectionScan должен быть таким, как этот

 MongoClient mongoClient = MongoClients.create();
 MongoDatabase database = mongoClient.getDatabase("localhost");
 Document commandResult = database.runCommand(new Document("parallelCollectionScan", "collectionName").append("numCursors", 3));

Ответ 4

Во-первых, как прокомментировал @xtreme-biker, производительность сильно зависит от вашего оборудования. В частности, первым моим советом будет проверка того, работаете ли вы на виртуальной машине или на собственном хосте. В моем случае с CentOS VM на i7 с накопителем SDD я могу читать 123 000 документов в секунду, но точно такой же код, который работает на хосте Windows на том же диске, считывает до 387 000 документов в секунду.

Затем давайте предположим, что вам действительно нужно прочитать полную коллекцию. Это означает, что вы должны выполнить полное сканирование. И давайте предположим, что вы не можете изменить конфигурацию своего сервера MongoDB, а только оптимизировать свой код.

Тогда все сводится к тому, что

collection.find().forEach((Block<Document>) document -> count.increment());

на самом деле.

Быстрое разворачивание MongoCollection.find() показывает, что на самом деле это делает:

ReadPreference readPref = ReadPreference.primary();
ReadConcern concern = ReadConcern.DEFAULT;
MongoNamespace ns = new MongoNamespace(databaseName,collectionName);
Decoder<Document> codec = new DocumentCodec();
FindOperation<Document> fop = new FindOperation<Document>(ns,codec);
ReadWriteBinding readBinding = new ClusterBinding(getCluster(), readPref, concern);
QueryBatchCursor<Document> cursor = (QueryBatchCursor<Document>) fop.execute(readBinding);
AtomicInteger count = new AtomicInteger(0);
try (MongoBatchCursorAdapter<Document> cursorAdapter = new MongoBatchCursorAdapter<Document>(cursor)) {
    while (cursorAdapter.hasNext()) {
        Document doc = cursorAdapter.next();
        count.incrementAndGet();
    }
}

Здесь FindOperation.execute() довольно быстро (до 10 мс), и большую часть времени тратится внутри цикла while и, в частности, внутри частного метода QueryBatchCursor.getMore()

getMore() вызывает DefaultServerConnection.command() и время расходуется в основном на две операции: 1) выборка строковых данных с сервера и 2) преобразование строковых данных в BsonDocument.

Оказывается, Mongo довольно умен в отношении того, сколько круговых поездок по сети это сделает, чтобы получить большой результирующий набор. Сначала он получит 100 результатов с помощью команды firstBatch, а затем выберет большие партии, а nextBatch будет размером партии в зависимости от размера коллекции до предела.

Итак, под деревом что-то вроде этого произойдет, чтобы получить первую партию.

ReadPreference readPref = ReadPreference.primary();
ReadConcern concern = ReadConcern.DEFAULT;
MongoNamespace ns = new MongoNamespace(databaseName,collectionName);
FieldNameValidator noOpValidator = new NoOpFieldNameValidator();
DocumentCodec payloadDecoder = new DocumentCodec();
Constructor<CodecProvider> providerConstructor = (Constructor<CodecProvider>) Class.forName("com.mongodb.operation.CommandResultCodecProvider").getDeclaredConstructor(Decoder.class, List.class);
providerConstructor.setAccessible(true);
CodecProvider firstBatchProvider = providerConstructor.newInstance(payloadDecoder, Collections.singletonList("firstBatch"));
CodecProvider nextBatchProvider = providerConstructor.newInstance(payloadDecoder, Collections.singletonList("nextBatch"));
Codec<BsonDocument> firstBatchCodec = fromProviders(Collections.singletonList(firstBatchProvider)).get(BsonDocument.class);
Codec<BsonDocument> nextBatchCodec = fromProviders(Collections.singletonList(nextBatchProvider)).get(BsonDocument.class);
ReadWriteBinding readBinding = new ClusterBinding(getCluster(), readPref, concern);
BsonDocument find = new BsonDocument("find", new BsonString(collectionName));
Connection conn = readBinding.getReadConnectionSource().getConnection();

BsonDocument results = conn.command(databaseName,find,noOpValidator,readPref,firstBatchCodec,readBinding.getReadConnectionSource().getSessionContext(), true, null, null);
BsonDocument cursor = results.getDocument("cursor");
long cursorId = cursor.getInt64("id").longValue();

BsonArray firstBatch = cursor.getArray("firstBatch");

Затем cursorId используется для извлечения каждой следующей партии.

На мой взгляд, "проблема" с реализацией драйвера заключается в том, что декодер String to JSON вводится, но JsonReader, в котором метод decode() полагается, - нет. Таким образом, даже до com.mongodb.internal.connection.InternalStreamConnection где вы уже находитесь рядом со связью сокета.

Поэтому я думаю, что вряд ли что-нибудь, что вы могли бы сделать, чтобы улучшить MongoCollection.find() если вы не сделаете так глубоко, как InternalStreamConnection.sendAndReceiveAsync()

Вы не можете уменьшить количество круговых поездок, и вы не можете изменить способ преобразования ответа в BsonDocument. Не обойтись без водителя и писать своего клиента, и я сомневаюсь, что это хорошая идея.

PD Если вы хотите попробовать часть кода выше, вам понадобится метод getCluster(), который требует грязного взлома в mongo-java-драйвер.

private Cluster getCluster() {
    Field cluster, delegate;
    Cluster mongoCluster = null;
    try {
        delegate = mongoClient.getClass().getDeclaredField("delegate");
        delegate.setAccessible(true);
        Object clientDelegate = delegate.get(mongoClient);
        cluster = clientDelegate.getClass().getDeclaredField("cluster");
        cluster.setAccessible(true);
        mongoCluster = (Cluster) cluster.get(clientDelegate);
    } catch (NoSuchFieldException | SecurityException | IllegalArgumentException | IllegalAccessException e) {
        System.err.println(e.getClass().getName()+" "+e.getMessage());
    }
    return mongoCluster;
}

Ответ 5

По моим подсчетам вы обрабатываете около 50 мегабайт/с (250 тыс. Строк/сек * 0,2 килобайта в секунду). Это касается как дискового и сетевого узких мест. Какое хранилище используется MongoDB? Какая ширина полосы пропускания между клиентом и сервером MongoDB? Пробовали ли вы совместное размещение сервера и клиента в высокоскоростной сети (> = 10 Гбит/с) с минимальной задержкой (<1,0 мс)? Имейте в виду, что если вы используете провайдера облачных вычислений, таких как AWS или GCP, у них будут узкие места виртуализации, которые находятся поверх физических.

Вы спросили о настройках, которые могут помочь. Вы можете попробовать изменить параметры сжатия в соединении и в коллекции (опции "none", snappy и zlib). Даже если ни улучшить snappy, видя разницу, что установка марки (или не делает), может помочь выяснить, какая часть системы находится под самой стрессе.

Java не имеет хорошей производительности для хрустания чисел по сравнению с C++ или Python, поэтому вы можете рассмотреть возможность перезаписи этой конкретной операции на одном из этих языков, а затем интегрировать ее с вашим Java-кодом. Я предлагаю вам выполнить тестовый прогон, просто перебирая данные в Python и сравнивая это с тем же в Java.