빠른 시작

전제 조건

PyMongoArrow 배포판 이(가) 설치되어 있는지 확인합니다. Python shell 에서 다음 항목이 예외 없이 실행 되어야 합니다.

>>> import pymongoarrow as pma

또한 이 튜토리얼에서는 MongoDB 인스턴스가 기본 호스트 및 포트에서 실행 중이라고 가정합니다. MongoDB를 다운로드하여 설치 한 후에는 다음 코드 예제와 같이 시작할 수 있습니다.

$ 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']}])

스키마 정의

PyMongoArrow는 데이터 스키마를 사용하여 쿼리 결과 세트를 표 형식으로 마셜링합니다. 이 스키마를 제공하지 않으면 PyMongoArrow가 데이터에서 스키마를 추론합니다. 다음 예와 같이 Schema 객체를 만들고 필드 이름을 유형 지정자에 매핑하여 스키마를 정의할 수 있습니다.

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

MongoDB는 내장된 문서를 사용하여 중첩된 데이터를 표현합니다. PyMongoArrow는 이러한 문서에 대해 최고 수준의 지원을 제공합니다.

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

PyMongoArrow는 목록과 중첩 목록도 지원합니다.

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

작업 찾기

다음 코드 예시에서는 amount 필드의 값이 0이 아닌 모든 레코드를 pandas.DataFrame 객체로 로드하는 방법을 보여 줍니다.

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

pyarrow.Table 인스턴스와 동일한 결과 집합을 로드할 수도 있습니다.

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

또는 polars.DataFrame 인스턴스로 사용할 수 있습니다.

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

또는 NumPy arrays 객체로 사용할 수 있습니다.

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

NumPy를 사용할 때 반환 값은 키는 필드이고 값은 해당 numpy.ndarray 인스턴스인 딕셔너리입니다.

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

집계 작업

애그리게이션 작업을 실행하는 것은 찾기 작업을 실행하는 것과 비슷하지만 수행하는 데 일련의 작업이 필요합니다.

다음은 모든 _id 값이 함께 그룹화되고 해당 amount 값이 합산된 새 데이터 프레임을 출력하는 aggregate_pandas_all() 메서드의 간단한 예입니다.

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

내장된 문서에 대해 애그리게이션 작업을 실행할 수도 있습니다. 다음 예제에서는 중첩된 txn 필드에서 값을 풀고 각 값의 개수를 계산한 다음 결과를 내림차순으로 정렬된 NumPy ndarray 객체 목록으로 반환합니다.

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

MongoDB에 쓰기

write() 메서드를 사용하여 MongoDB에 다음 유형의 객체를 쓸 수 있습니다.

• 화살표 Table

• Pandas DataFrame

• NumPy ndarray

• 극지 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}

다른 형식으로 쓰기

결과 세트가 로드되면 패키지가 지원하는 모든 형식으로 결과를 작성할 수 있습니다.

예를 들어, 변수 arrow_table 가 참고 테이블을 example.parquet Parquet 파일에 쓰려면 다음 코드를 실행하세요.

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

Pandas는 또한 DataFrame 인스턴스를 CSV 및 HDF를 포함한 다양한 형식으로 쓸 수 있도록 지원합니다. 변수 df 가 참고하는 데이터 프레임을 out.csv 라는 이름의 CSV 파일에 쓰려면 다음 코드를 실행하세요.

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

폴라 API는 앞의 두 가지 예를 혼합한 것입니다.

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