¿Cómo importar datos de mongodb a pandas?

99

Tengo una gran cantidad de datos en una colección en mongodb que necesito analizar. ¿Cómo importo esos datos a pandas?

Soy nuevo en pandas y numpy.

EDITAR: La colección mongodb contiene valores de sensor etiquetados con fecha y hora. Los valores del sensor son del tipo de datos flotantes.

Data de muestra:

{
"_cls" : "SensorReport",
"_id" : ObjectId("515a963b78f6a035d9fa531b"),
"_types" : [
    "SensorReport"
],
"Readings" : [
    {
        "a" : 0.958069536790466,
        "_types" : [
            "Reading"
        ],
        "ReadingUpdatedDate" : ISODate("2013-04-02T08:26:35.297Z"),
        "b" : 6.296118156595,
        "_cls" : "Reading"
    },
    {
        "a" : 0.95574014778624,
        "_types" : [
            "Reading"
        ],
        "ReadingUpdatedDate" : ISODate("2013-04-02T08:27:09.963Z"),
        "b" : 6.29651468650064,
        "_cls" : "Reading"
    },
    {
        "a" : 0.953648289182713,
        "_types" : [
            "Reading"
        ],
        "ReadingUpdatedDate" : ISODate("2013-04-02T08:27:37.545Z"),
        "b" : 7.29679823731148,
        "_cls" : "Reading"
    },
    {
        "a" : 0.955931884300997,
        "_types" : [
            "Reading"
        ],
        "ReadingUpdatedDate" : ISODate("2013-04-02T08:28:21.369Z"),
        "b" : 6.29642922525632,
        "_cls" : "Reading"
    },
    {
        "a" : 0.95821381,
        "_types" : [
            "Reading"
        ],
        "ReadingUpdatedDate" : ISODate("2013-04-02T08:41:20.801Z"),
        "b" : 7.28956613,
        "_cls" : "Reading"
    },
    {
        "a" : 4.95821335,
        "_types" : [
            "Reading"
        ],
        "ReadingUpdatedDate" : ISODate("2013-04-02T08:41:36.931Z"),
        "b" : 6.28956574,
        "_cls" : "Reading"
    },
    {
        "a" : 9.95821341,
        "_types" : [
            "Reading"
        ],
        "ReadingUpdatedDate" : ISODate("2013-04-02T08:42:09.971Z"),
        "b" : 0.28956488,
        "_cls" : "Reading"
    },
    {
        "a" : 1.95667927,
        "_types" : [
            "Reading"
        ],
        "ReadingUpdatedDate" : ISODate("2013-04-02T08:43:55.463Z"),
        "b" : 0.29115237,
        "_cls" : "Reading"
    }
],
"latestReportTime" : ISODate("2013-04-02T08:43:55.463Z"),
"sensorName" : "56847890-0",
"reportCount" : 8
}
Nithin
fuente
El uso de un tipo de campo personalizado con MongoEngine puede hacer que almacenar y recuperar Pandas DataFrames sea tan simple comomongo_doc.data_frame = my_pandas_df
Jthorpe

Respuestas:

133

pymongo podría echarte una mano, los siguientes son algunos códigos que estoy usando:

import pandas as pd
from pymongo import MongoClient


def _connect_mongo(host, port, username, password, db):
    """ A util for making a connection to mongo """

    if username and password:
        mongo_uri = 'mongodb://%s:%s@%s:%s/%s' % (username, password, host, port, db)
        conn = MongoClient(mongo_uri)
    else:
        conn = MongoClient(host, port)


    return conn[db]


def read_mongo(db, collection, query={}, host='localhost', port=27017, username=None, password=None, no_id=True):
    """ Read from Mongo and Store into DataFrame """

    # Connect to MongoDB
    db = _connect_mongo(host=host, port=port, username=username, password=password, db=db)

    # Make a query to the specific DB and Collection
    cursor = db[collection].find(query)

    # Expand the cursor and construct the DataFrame
    df =  pd.DataFrame(list(cursor))

    # Delete the _id
    if no_id:
        del df['_id']

    return df
esperando
fuente
Gracias, este es el método que terminé usando. También tenía una serie de documentos incrustados en cada fila. Así que tuve que repetir eso también dentro de cada fila. ¿¿Hay una mejor manera de hacer esto??
Nithin
¿Es posible proporcionar algunas muestras de la estructura de su mongodb?
esperandokuo
3
Tenga list()en cuenta que el interior se df = pd.DataFrame(list(cursor))evalúa como una lista o generador, para mantener fría la CPU. Si tiene un trillón de elementos de datos, y las siguientes líneas los habrían dividido razonablemente, nivel de detalle y recortado, todo el shmegegge todavía es seguro para caer. Bien.
Phlip
2
Es muy lento @ df = pd.DataFrame(list(cursor)). La búsqueda pura de db es mucho más rápida. ¿Podríamos cambiar el listcasting a otra cosa?
Peter.k
1
@Peter esa línea también me llamó la atención. Lanzar un cursor de base de datos, que está diseñado para ser iterable y potencialmente envuelve grandes cantidades de datos, en una lista en memoria no me parece inteligente.
Rafa
41

Puede cargar sus datos de mongodb en pandas DataFrame usando este código. Esto funciona para mi. Ojalá para ti también.

import pymongo
import pandas as pd
from pymongo import MongoClient
client = MongoClient()
db = client.database_name
collection = db.collection_name
data = pd.DataFrame(list(collection.find()))
saimadhu.polamuri
fuente
24

Monaryhace exactamente eso, y es súper rápido . ( otro enlace )

Vea esta interesante publicación que incluye un tutorial rápido y algunos horarios.

shx2
fuente
¿Monary admite el tipo de datos de cadena?
Snehal Parmar
Probé Monary, pero está tomando mucho tiempo. ¿Me falta alguna optimización? Probado client = Monary(host, 27017, database="db_tmp") columns = ["col1", "col2"] data_type = ["int64", "int64"] arrays = client.query("db_tmp", "coll", {}, columns, data_type)para 50000registros lleva alrededor 200s.
nishant
Que los sonidos extremadamente lento ... Francamente, no sé cuál es el estado de este proyecto es, ahora, 4 años más tarde ...
shx2
16

Según PEP, lo simple es mejor que complicado:

import pandas as pd
df = pd.DataFrame.from_records(db.<database_name>.<collection_name>.find())

Puede incluir condiciones como lo haría trabajando con una base de datos mongoDB normal o incluso usar find_one () para obtener solo un elemento de la base de datos, etc.

¡y voilá!

Cy Bu
fuente
pd.DataFrame.from_records parece ser tan lento como DataFrame (list ()), pero los resultados son muy inconsistentes. %% del tiempo mostró algo entre 800 ms y 1,9 s
AFD
1
Esto no es bueno para registros grandes ya que no muestra un error de memoria, instread bloquea el sistema para datos demasiado grandes. mientras que pd.DataFrame (lista (cursor)) muestra error de memoria.
Amulya Acharya
13
import pandas as pd
from odo import odo

data = odo('mongodb://localhost/db::collection', pd.DataFrame)
fengwt
fuente
9

Para tratar con datos fuera del núcleo (que no encajan en la RAM) de manera eficiente (es decir, con ejecución paralela), puede probar el ecosistema Python Blaze : Blaze / Dask / Odo.

Blaze (y Odo ) tiene funciones listas para usar para lidiar con MongoDB.

Algunos artículos útiles para comenzar:

Y un artículo que muestra qué cosas asombrosas son posibles con la pila Blaze: Analizando 1.7 mil millones de comentarios de Reddit con Blaze e Impala (esencialmente, consultando 975 Gb de comentarios de Reddit en segundos).

PD: No estoy afiliado a ninguna de estas tecnologías.

Dennis Golomazov
fuente
1
También escribí una publicación usando Jupyter Notebook con un ejemplo de cómo Dask ayuda a acelerar la ejecución incluso en datos que se ajustan a la memoria mediante el uso de múltiples núcleos en una sola máquina.
Dennis Golomazov
8

Otra opción que encontré muy útil es:

from pandas.io.json import json_normalize

cursor = my_collection.find()
df = json_normalize(cursor)

de esta manera, obtiene el despliegue de documentos mongodb anidados de forma gratuita.

Ikar Pohorský
fuente
2
Recibí un error con este métodoTypeError: data argument can't be an iterator
Feria Gabriel
2
Extraño, esto funciona en mi pitón 3.6.7usando pandas 0.24.2. ¿Quizás puedas intentarlo en su df = json_normalize(list(cursor))lugar?
Ikar Pohorský
Para +1. docs, el argumento max_level define el nivel máximo de profundidad de dictado. Acabo de hacer una prueba y no es cierto, por lo que algunas columnas deberían dividirse con accesrors .str. Aún así, es una característica muy buena para trabajar con mongodb.
Mauricio Maroto
5

Utilizando

pandas.DataFrame(list(...))

consumirá mucha memoria si el resultado del iterador / generador es grande

mejor generar pequeños trozos y concat al final

def iterator2dataframes(iterator, chunk_size: int):
  """Turn an iterator into multiple small pandas.DataFrame

  This is a balance between memory and efficiency
  """
  records = []
  frames = []
  for i, record in enumerate(iterator):
    records.append(record)
    if i % chunk_size == chunk_size - 1:
      frames.append(pd.DataFrame(records))
      records = []
  if records:
    frames.append(pd.DataFrame(records))
  return pd.concat(frames)
Deo Leung
fuente
1

Siguiendo esta gran respuesta esperandokuo, me gustaría agregar la posibilidad de hacerlo usando chunksize en línea con .read_sql () y .read_csv () . Amplío la respuesta de Deu Leung evitando ir uno por uno a cada 'registro' del 'iterador' / 'cursor'. Tomaré prestada la función read_mongo anterior .

def read_mongo(db, 
           collection, query={}, 
           host='localhost', port=27017, 
           username=None, password=None,
           chunksize = 100, no_id=True):
""" Read from Mongo and Store into DataFrame """


# Connect to MongoDB
#db = _connect_mongo(host=host, port=port, username=username, password=password, db=db)
client = MongoClient(host=host, port=port)
# Make a query to the specific DB and Collection
db_aux = client[db]


# Some variables to create the chunks
skips_variable = range(0, db_aux[collection].find(query).count(), int(chunksize))
if len(skips_variable)<=1:
    skips_variable = [0,len(skips_variable)]

# Iteration to create the dataframe in chunks.
for i in range(1,len(skips_variable)):

    # Expand the cursor and construct the DataFrame
    #df_aux =pd.DataFrame(list(cursor_aux[skips_variable[i-1]:skips_variable[i]]))
    df_aux =pd.DataFrame(list(db_aux[collection].find(query)[skips_variable[i-1]:skips_variable[i]]))

    if no_id:
        del df_aux['_id']

    # Concatenate the chunks into a unique df
    if 'df' not in locals():
        df =  df_aux
    else:
        df = pd.concat([df, df_aux], ignore_index=True)

return df
Rafael Valero
fuente
1

Un enfoque similar como Rafael Valero, waitkuo y Deu Leung usando paginación :

def read_mongo(
       # db, 
       collection, query=None, 
       # host='localhost', port=27017, username=None, password=None,
       chunksize = 100, page_num=1, no_id=True):

    # Connect to MongoDB
    db = _connect_mongo(host=host, port=port, username=username, password=password, db=db)

    # Calculate number of documents to skip
    skips = chunksize * (page_num - 1)

    # Sorry, this is in spanish
    # https://www.toptal.com/python/c%C3%B3digo-buggy-python-los-10-errores-m%C3%A1s-comunes-que-cometen-los-desarrolladores-python/es
    if not query:
        query = {}

    # Make a query to the specific DB and Collection
    cursor = db[collection].find(query).skip(skips).limit(chunksize)

    # Expand the cursor and construct the DataFrame
    df =  pd.DataFrame(list(cursor))

    # Delete the _id
    if no_id:
        del df['_id']

    return df
Jordy Cuan
fuente
0

Puedes lograr lo que quieras con pdmongo en tres líneas:

import pdmongo as pdm
import pandas as pd
df = pdm.read_mongo("MyCollection", [], "mongodb://localhost:27017/mydb")

Si sus datos son muy grandes, puede hacer una consulta agregada primero filtrando los datos que no desea y luego mapeándolos a las columnas deseadas.

A continuación, se muestra un ejemplo de asignación Readings.aa columna ay filtrado por reportCountcolumna:

import pdmongo as pdm
import pandas as pd
df = pdm.read_mongo("MyCollection", [{'$match': {'reportCount': {'$gt': 6}}}, {'$unwind': '$Readings'}, {'$project': {'a': '$Readings.a'}}], "mongodb://localhost:27017/mydb")

read_mongoacepta los mismos argumentos que el agregado de pymongo

pakallis
fuente