java - que - mongodb español

MongoDB Java API rendimiento de lectura lenta

Question

Estamos leyendo de un MongoDB local todos los documentos de una colección y el rendimiento no es muy brillante.

Necesitamos volcar todos los datos, no se preocupe por eso, solo confíe en que es realmente necesario y que no hay solución posible.

Tenemos 4mio documentos que se parecen a:

{
    "_id":"4d094f58c96767d7a0099d49",
    "exchange":"NASDAQ",
    "stock_symbol":"AACC",
    "date":"2008-03-07",
    "open":8.4,
    "high":8.75,
    "low":8.08,
    "close":8.55,
    "volume":275800,
    "adj close":8.55
}

Y estamos usando esto por ahora código trivial para leer:

MongoClient mongoClient = MongoClients.create();
MongoDatabase database = mongoClient.getDatabase("localhost");
MongoCollection<Document> collection = database.getCollection("test");

MutableInt count = new MutableInt();
long start = System.currentTimeMillis();
collection.find().forEach((Block<Document>) document -> count.increment() /* actually something more complicated */ );
long start = System.currentTimeMillis();

Estamos leyendo toda la colección a 16 segundos (250k fila / seg), eso no es realmente impresionante con documentos pequeños. Ten en cuenta que queremos cargar 800mio filas. Ningún agregado, mapa reducido o similar son posibles.

¿Es esto tan rápido como MongoDB obtiene o hay otras maneras de cargar documentos más rápido (otras técnicas, mover Linux, más RAM, configuraciones ...)?

Answer 1

Lo que creo que debí hacer en su caso fue una solución simple y, al mismo tiempo, una forma eficiente es maximizar el rendimiento general mediante el uso de la colección de exploración paralela

Permite que las aplicaciones usen múltiples cursores paralelos cuando leen todos los documentos de una colección, lo que aumenta el rendimiento. El comando parallelCollectionScan devuelve un documento que contiene una matriz de información del cursor.

Cada cursor proporciona acceso a la devolución de un conjunto parcial de documentos de una colección. Iterando cada cursor devuelve todos los documentos de la colección. Los cursores no contienen los resultados del comando de base de datos. El resultado del comando de la base de datos identifica los cursores, pero no contiene ni constituye los cursores.

Un ejemplo simple con parallelCollectionScan debería ser algo como este

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

Answer 2

No especificó su caso de uso, por lo que es muy difícil decirle cómo ajustar su consulta. (Por ejemplo, ¿quién querría cargar 800 mil filas por vez solo para el recuento?).

Dado su esquema, creo que sus datos son casi de solo lectura y su tarea está relacionada con la agregación de datos.

Su trabajo actual es simplemente leer los datos (lo más probable es que su conductor lea en lotes), luego se detenga, luego realice algunos cálculos (demonios, sí, se usa un envoltorio int para aumentar el tiempo de procesamiento), y luego repita. Eso no es un buen enfoque. El DB no es mágicamente rápido si no lo accedes de la manera correcta.

Si el cálculo no es demasiado complejo, le sugiero que utilice el marco de agregación en lugar de cargar todo en su RAM.

Algo que debes considerar para mejorar tu agregación:

1. Divide tu conjunto de datos en un conjunto más pequeño. (Por ejemplo: Partición por date , partición por exchange ...). Agregue un índice para admitir esa partición y opere la agregación en la partición y luego combine el resultado (enfoque típico de divide-n-conquista)
2. Proyecto solo campos necesarios
3. Filtrar el documento innecesario (si es posible)
4. Permitir uso de disco si no puede realizar su agregación en la memoria (si alcanza el límite de 100 MB por pipilina).
5. Use una tubería integrada para acelerar su cálculo (por ejemplo: $count para su ejemplo)

Si su cálculo es demasiado complejo que no puede expresar con el marco de agregación, use mapReduce . Funciona en el proceso mongod y los datos no necesitan transferirse a través de la red a su memoria.

Actualizado

Así que parece que quieres hacer un procesamiento OLAP y te quedaste atascado en el paso ETL.

No es necesario y debe evitar cargar todos los datos OLTP completos en OLAP cada vez. Solo necesitas cargar nuevos cambios en tu almacén de datos. Entonces, la primera carga / descarga de datos lleva más tiempo, es normal y aceptable.

Para la primera carga, debes considerar los siguientes puntos:

1. Divide-N-Conquer, de nuevo, divide sus datos en conjuntos de datos más pequeños (con predicado como fecha / intercambio / etiqueta de stock ...)
2. Haga cálculos paralelos, luego combine su resultado (debe particionar su conjunto de datos correctamente)
3. forEach en lote en lugar de procesar en forEach : cargue la partición de datos y luego calcule en lugar de computar uno por uno.

Answer 3

Primero, como comentó @ xtreme-biker, el rendimiento depende en gran medida de su hardware. Específicamente, mi primer consejo sería verificar si está ejecutando en una máquina virtual o en un host nativo. En mi caso con una máquina virtual CentOS en un i7 con una unidad SDD puedo leer 123,000 documentos por segundo, pero exactamente el mismo código que se ejecuta en Windows Host en la misma unidad lee hasta 387,000 documentos por segundo.

A continuación, asumamos que realmente necesitas leer la colección completa. Esto quiere decir que debe realizar un escaneo completo. Y supongamos que no puede cambiar la configuración de su servidor MongoDB, sino solo optimizar su código.

Entonces todo se reduce a lo que

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

en realidad lo hace.

Un rápido desenrollado de MongoCollection.find () muestra que realmente hace esto:

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

Aquí FindOperation.execute() es bastante rápido (menos de 10 ms) y la mayor parte del tiempo se pasa dentro del bucle while, y específicamente dentro del método privado QueryBatchCursor.getMore()

getMore() llama a DefaultServerConnection.command() y su tiempo se consume básicamente en dos operaciones: 1) buscar datos de cadena desde el servidor y 2) convertir datos de cadena en BsonDocument.

Resulta que Mongo es bastante inteligente con respecto a la cantidad de viajes de red que realizará para obtener un gran conjunto de resultados. Primero obtendrá 100 resultados con un comando firstBatch y luego obtendrá lotes más grandes con nextBatch siendo el tamaño del lote dependiendo del tamaño de la colección hasta un límite.

Entonces, debajo de la madera, algo así sucederá para obtener el primer lote.

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

Luego, el cursorId se utiliza para obtener cada lote siguiente.

En mi opinión, el "problema" con la implementación del controlador es que el decodificador String to JSON se inyecta, pero el JsonReader, en el que se basa el método decode (), no lo está. Esto es así incluso en com.mongodb.internal.connection.InternalStreamConnection donde ya está cerca de la comunicación de socket.

Por lo tanto, creo que casi no hay nada que puedas hacer para mejorar MongoCollection.find() menos que profundices tanto como InternalStreamConnection.sendAndReceiveAsync()

No puede reducir el número de viajes de ida y vuelta y no puede cambiar la forma en que la respuesta se convierte en BsonDocument. No sin pasar por alto el controlador y escribir a tu propio cliente, lo cual dudo es una buena idea.

PD Si desea probar algunos de los códigos anteriores, necesitará el método getCluster () que requiere un truco sucio en mongo-java-driver .

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

Answer 4

Según mi conteo, estás procesando unos 50 MiB / s (250k row / sec * 0.2 KiB / row). Eso está llegando tanto a la unidad de disco como al territorio del cuello de botella de la red. ¿Qué tipo de almacenamiento utiliza MongoDB? ¿Qué tipo de ancho de banda tiene entre el cliente y el servidor MongoDB? ¿Ha intentado ubicar conjuntamente el servidor y el cliente en una red de alta velocidad (> = 10 Gib / s) con una latencia mínima (<1.0 ms)? Tenga en cuenta que si utiliza un proveedor de computación en la nube como AWS o GCP, tendrán cuellos de botella en la virtualización que están por encima de los físicos.

Usted preguntó acerca de la configuración que podría ayudar. Puede intentar cambiar la configuración de compresión en la connection y en la collection (las opciones son "none", snappy y zlib ). Incluso si ninguno de los dos mejora snappy , ver la diferencia que hace (o no hace) la configuración podría ayudar a determinar qué parte del sistema está bajo más estrés.

Java no tiene un buen rendimiento para el procesamiento de números en comparación con C ++ o Python, por lo que puede considerar volver a escribir esta operación en particular en uno de esos lenguajes y luego integrarla con su código Java. Le sugiero que haga una prueba de funcionamiento de solo hacer un bucle sobre los datos en Python y compararlo con el mismo en Java.

Answer 5

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

Esta línea puede acumular mucho tiempo ya que está iterando más de 250k registros en la memoria.

Para comprobar rápidamente si ese es el caso, puede intentar esto:

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

Esto le ayudará a comprender cuánto tiempo lleva en realidad obtener los documentos de la base de datos y cuánto tiempo lleva la iteración.