Início rápido

Pré-requisitos

Verifique se você tem a distribuição PyMongoArrow instalada. Na shell do Python, o seguinte deve ser executado sem gerar uma exceção:

>>> import pymongoarrow as pma

Este tutorial também pressupõe que uma instância MongoDB esteja sendo executada no host e na porta padrão. Depois de baixar e instalar o MongoDB, você pode iniciá-lo conforme mostrado no exemplo de código a seguir:

$ mongod

from pymongoarrow.monkey import patch_all
patch_all()

from datetime import datetime
from pymongo import MongoClient
client = MongoClient()
client.db.data.insert_many([
  {'_id': 1, 'amount': 21, 'last_updated': datetime(2020, 12, 10, 1, 3, 1), 'account': {'name': 'Customer1', 'account_number': 1}, 'txns': ['A']},
  {'_id': 2, 'amount': 16, 'last_updated': datetime(2020, 7, 23, 6, 7, 11), 'account': {'name': 'Customer2', 'account_number': 2}, 'txns': ['A', 'B']},
  {'_id': 3, 'amount': 3,  'last_updated': datetime(2021, 3, 10, 18, 43, 9), 'account': {'name': 'Customer3', 'account_number': 3}, 'txns': ['A', 'B', 'C']},
  {'_id': 4, 'amount': 0,  'last_updated': datetime(2021, 2, 25, 3, 50, 31), 'account': {'name': 'Customer4', 'account_number': 4}, 'txns': ['A', 'B', 'C', 'D']}])

Definindo o esquema

O PyMongoArrow depende de um esquema de dados para organizar conjuntos de resultados de query em formato tabular. If you don't provide this schema, PyMongoArrow infers one from the data. Você pode definir o esquema criando um objeto Schema e mapeando os nomes dos campos para especificadores de tipo, conforme mostrado no exemplo a seguir:

from pymongoarrow.api import Schema
schema = Schema({'_id': int, 'amount': float, 'last_updated': datetime})

O MongoDB usa documentos incorporados para representar dados aninhados. O PyMongoArrow oferece suporte de primeira classe para estes documentos:

schema = Schema({'_id': int, 'amount': float, 'account': { 'name': str, 'account_number': int}})

O PyMongoArrow também suporta listas e listas aninhadas:

from pyarrow import list_, string
schema = Schema({'txns': list_(string())})
polars_df = client.db.data.find_polars_all({'amount': {'$gt': 0}}, schema=schema)

Encontrar operações

O seguinte exemplo de código mostra como carregar todos os registros que têm um valor diferente de zero para o campo amount como um objeto pandas.DataFrame :

df = client.db.data.find_pandas_all({'amount': {'$gt': 0}}, schema=schema)

Você também pode carregar o mesmo conjunto de resultados que uma instância do pyarrow.Table :

arrow_table = client.db.data.find_arrow_all({'amount': {'$gt': 0}}, schema=schema)

Ou como uma instância polars.DataFrame :

df = client.db.data.find_polars_all({'amount': {'$gt': 0}}, schema=schema)

Ou como um objeto NumPy arrays :

ndarrays = client.db.data.find_numpy_all({'amount': {'$gt': 0}}, schema=schema)

Ao usar o NumPy, o valor retornado é um dicionário onde as chaves são nomes de campos e os valores são as instâncias numpy.ndarray correspondentes.

arrow_table = client.db.data.find_arrow_all({'amount': {'$gt': 0}})

Operações agregadas

A execução de uma operação agregada é semelhante à execução de uma operação de busca, mas é necessária uma sequência de operações para ser executada.

O seguinte é um exemplo simples do método aggregate_pandas_all() que gera um novo dataframe no qual todos os valores _id são agrupados e seus valores amount somados:

df = client.db.data.aggregate_pandas_all([{'$group': {'_id': None, 'total_amount': { '$sum': '$amount' }}}])

Você também pode executar operações agregadas em documentos incorporados. O exemplo a seguir desenrola os valores no campo txn aninhado, conta o número de cada valor e retorna os resultados como uma lista de objetos NumPy ndarray , classificados em ordem decrescente:

pipeline = [{'$unwind': '$txns'}, {'$group': {'_id': '$txns', 'count': {'$sum': 1}}}, {'$sort': {"count": -1}}]
ndarrays = client.db.data.aggregate_numpy_all(pipeline)

Escrevendo no MongoDB

Você pode utilizar o método write() para gravar objetos dos seguintes tipos no MongoDB:

• Seta Table

• Pandas DataFrame

• NumPy ndarray

• Áries DataFrame

from pymongoarrow.api import write
from pymongo import MongoClient
coll = MongoClient().db.my_collection
write(coll, df)
write(coll, arrow_table)
write(coll, ndarrays)

data_a = [1, 2, 3]
data_b = [1, None, 3]
data = Table.from_pydict(
   {
      "a": data_a,
      "b": data_b,
   },
)
coll.drop()
write(coll, data)
col_data = list(coll.find({}))
coll.drop()
write(coll, data, exclude_none=True)
col_data_exclude_none = list(coll.find({}))
print(col_data)
print(col_data_exclude_none)

{'_id': ObjectId('...'), 'a': 1, 'b': 1}
{'_id': ObjectId('...'), 'a': 2, 'b': None}
{'_id': ObjectId('...'), 'a': 3, 'b': 3}
{'_id': ObjectId('...'), 'a': 1, 'b': 1}
{'_id': ObjectId('...'), 'a': 2}
{'_id': ObjectId('...'), 'a': 3, 'b': 3}

Escrevendo para outros formatos

Depois que os conjuntos de resultados forem carregados, você poderá escrevê-los em qualquer formato suportado pelo pacote.

Por exemplo, para gravar a tabela referenciada pela variável arrow_table em um arquivo Parquet denominado example.parquet , execute o seguinte código:

import pyarrow.parquet as pq
pq.write_table(arrow_table, 'example.parquet')

O Pandas também suporta a gravação de instâncias do DataFrame em uma variedade de formatos, incluindo CSV e alta definição. Para escrever o período de dados referenciado pela variável df em um arquivo CSV denominado out.csv, execute o seguinte código:

df.to_csv('out.csv', index=False)

A Polars API é uma mistura dos dois exemplos anteriores:

import polars as pl
df = pl.DataFrame({"foo": [1, 2, 3, 4, 5]})
df.write_parquet('example.parquet')