問題タブ [pymongo-3.x]

ここに私のサンプルデータがあります:

}

平均値を 2 時間間隔でグループ化し、次の集計クエリを作成しようとしています。

これによりエラーが発生し A pipeline stage specification object must contain exactly one field.、すべての調査を行いましたが、目的の結果が得られません。

約 3,000 万のドキュメントを含む MongoDB データベースがあります。単一のマルチプロセッサ コンピューター (144 GB RAM、CentOS 7.1 を実行) から提供され、シャーディングされていません。Python 3.4 を使用して、pymongo 3.2.2 でクエリを実行します。データベースはあまり使用されておらず、少数のライターが現在レコードを更新しています。findが一貫した結果を返さない状況は次のとおりです。

複数回実行すると、このコードは 72 を出力することもあれば、はるかに大きな数 (2874400 を超える) を出力することもあります。数値が大きいほど常に同じとは限りません。私のデータから知る限り、数値が大きいほど正しい可能性が高くなります。それでも、72 を出力することがある理由を説明するには途方に暮れています。

この一貫性のない動作の潜在的な原因は何ですか?

(注: はい、count(...)一致するドキュメントの数をカウントするために演算子を使用できることはわかっています。上記のコードは、一致するドキュメントをカウントする効率的な方法を提示するためではなく、矛盾を説明するためのフラグメントです。)

asyncio に関する質問に戻ってきました。私はそれが非常に便利だと思います (特にスレッドを使用した GIL のおかげで)。いくつかのコードのパフォーマンスを向上させようとしています。

私のアプリケーションは次のことを行っています:

• 1 バックグラウンド デーモン スレッド「A」は、接続されたクライアントからイベントを受信し、SetQueue (重複 ID を削除する単純なイベント キュー) に入力し、DB に挿入することによって反応します。このデーモンは別のモジュールから取得します (基本的には、イベントが受信されたときからのコールバックを制御します)。以下のサンプル コードでは、これを私が生成したスレッドに置き換えました。これは、キューに 20 個のアイテムを入力し、終了する前に DB 挿入を模倣するだけです。

• 1 バックグラウンド デーモン スレッド "B" が起動され (loop_start)、次のコルーチンの実行が完了するまでループします。

  • キュー内のすべてのアイテムを取得します (空でない場合、x 秒間コントロールを解放してから、コルーチンを再起動します)

  • キュー内の ID ごとに、次のような連鎖コルーチンを起動します。

    • DB からその ID に関連するすべての情報を取得するだけのタスクを作成して待機します。asyncio をサポートする MotorClient を使用して、タスク自体で await を実行しています。

    • Pool of Processes executor を使用して、ID ごとにプロセスを起動し、DB データを使用して CPU を集中的に処理します。

• メイン スレッドは db_client を初期化し、loop_start コマンドと stop コマンドを受け取るだけです。

それは基本的にそれです。

今、私は可能な限りパフォーマンスを向上させようとしています。

私の現在の問題はmotor.motor_asyncio.AsyncioMotorClient()、このように使用することです:

1. メインスレッドで初期化され、そこでインデックスを作成したい
2. スレッド「A」は DB 挿入を実行する必要があります
3. スレッド「B」は DB の検索/読み取りを実行する必要があります

これどうやってするの？Motor は、明らかに単一のイベントループを使用する単一スレッドアプリケーション向けであると述べています。ここで、スレッド「A」とスレッド「B」に 1 つずつ、合計 2 つのイベント ループを作成する必要があることに気付きました。これは最適ではありませんが、同じ動作を維持しながら call_soon_threadsafe で単一のイベントループを使用することはできませんでした...そして、ギルバウンドのCPUコアの制御を解放する2つのイベントループでパフォーマンスが向上していると思います.

3 つの異なる AsyncioMotorClient インスタンス (スレッドごとに 1 つ) を使用し、上記のように使用する必要がありますか? 試行中にさまざまなエラーで失敗しました。

これは、Asynchro の MotorClient 初期化だけを含まない私のサンプル コードです。__init__

MongoDB レプリカ セットに接続すると、セカンダリから読み取ることができますが、その後、同じ接続を使用して書き込むと、次のエラーが表示されます。

Python 3.4.3 ピモンゴ 3.2.2

私の接続：

重要:「replicaSet」フィールドと「readPrefence」フィールドの両方を削除すると、接続が機能します。

サーバー内:

デシベル.isMaster()

モンゴッド0

モンゴッド1

モンゴッド2

私は現在、分散コンピューティングに取り組んでいます。私のワーカーは結果を mongoDB データベースに挿入して返します。コードは正常に動作しますが、接続が開いたままになり、システムのソケットが不足する瞬間があります。これが私のワーカーコードです：

このコマンド「netstat -anbo」を使用すると、まだ開いているすべてのソケット (3000 以上) を確認できます。ワーカーの最大数は 14 ですが、10 000 以上のタスクを処理する必要があります。

タイムアウトを設定しようとしましたが、効果がありません。

データベースを再起動せずにソケットを閉じるにはどうすればよいですか?

Python 2.7.12 ピモンゴ 3.3 mongoDB 3.2.10

pymongoを使用してmongodbで検証をスキップすることは可能ですか?

pymongo から以下のクエリを実行したい:

どんな助けでも大歓迎です。

ありがとう、アシュウィン。