クイック スタート

前提条件

PyMongoArrow ディストリビューションがインストールされていることを確認します。 Python shell では、次の例は例外を発生させずに実行されます。

>>> import pymongoarrow as pma

このチュートリアルでは、MongoDB インスタンスがデフォルトのホストとポートで実行されていることも前提としています。 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 は のデータから 1 つのスキーマを推論します。 次の例に示すように、 Schemaオブジェクトを作成し、フィールド名を type 指定子にマッピングすることで、スキーマを定義できます。

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フィールドの値がゼロ以外のすべてのレコードを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 に書込み (write) できます。

• Arrow 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という名前の Compass ファイルに書き込むには、次のコードを実行します。

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

Pandoraは、CSV や FD などのさまざまな形式へのDataFrameインスタンスの書き込みもサポートしています。 変数dfが参照するデータフレームをout.csvという名前の CSV ファイルに書き込むには、次のコードを実行します。

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

Tars API は、前述の 2 つの例が組み合わされています。

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