問題タブ [dask]

TL;DR : 分散読み取りからメタデータ (解析中のエラー) を dask データフレーム コレクションに収集するにはどうすればよいですか。

私は現在、dask.DataFrame にフィードするために使用している独自のファイル形式を持っています。ファイルパスを受け入れて pandas.DataFrame を返す関数があります。これは、複数のファイルを同じ dask.DataFrame に正常にロードするために dask.DataFrame によって内部的に使用されます。

最近まで、私は自分のコードを使用して複数の pandas.DataFrames を 1 つにマージしていましたが、現在は代わりに dask を使用することに取り組んでいます。ファイル形式を解析するときに、エラーや特定の条件が発生する場合があります。ログに記録し、dask.DataFrame オブジェクトにメタデータ (ログ、データの起源など) として関連付けたいと考えています。

合理的な場合、MultiImdices をかなり頻繁に使用していることに注意することが重要です (13 のインデックス レベル、3 列のレベル)。特定の行ではなくデータフレーム全体を記述するメタデータについては、属性を使用しています。

カスタム関数を使用して、実際の DataFrame を含むタプルでメタデータを渡すことができました。pandas を使用すると、それを _metadata フィールドに追加し、DataFrame オブジェクトの属性として追加できます。dask フレームワークを使用しているときに、別の pandas.DataFrame オブジェクトからメタデータを収集するにはどうすればよいですか?

ありがとう！

私はdaskカスタムグラフを使用しています

グラフ定義を印刷するにはどうすればよいですか

たとえば、このグラフを考えてみましょう

上記の説明を次のように呼び出して印刷したいと思います

dot_graph(dask,'image.png') については知っていますが、テキストによる説明を取得したい

かなり大きな csv ファイル (〜 10 GB) がいくつかあり、分析に dask を利用したいと考えています。ただし、dask オブジェクトを読み込むように設定したパーティションの数によっては、groupby の結果が変わります。私の理解では、dask はアウトオブコア処理の利点のためにパーティションを利用しましたが、それでも適切な groupby 出力が返されるということでした。これは当てはまらないようで、どの代替設定が必要かを判断するのに苦労しています. 以下に小さな例を示します。

groupby ステップは、データフレーム全体を調べるのではなく、各パーティション内でのみ動作しますか? この場合、npartitions=1 を設定するのは簡単で、パフォーマンスに大きな影響を与えるようには見えませんが、read_csv は特定の数のパーティションを自動的に設定するため、groupby の結果が正確であることを確認するには、どのように呼び出しを設定すればよいでしょうか?

ありがとう！

フィルターする画像スタックをn持つ h5 ファイルを含むディレクトリがあります。m画像ごとに、dask並列配列を使用してフィルタリング (ガウスおよびラプラシアン) を実行し、処理を高速化します ( Ref to Dask )。apply_parallel()scikit-imageの関数で dask 配列を使用します 。
を使用して小さなサーバーで処理を実行します20 cpus。

どの並列戦略を使用するのがより理にかなっているのか、アドバイスをもらいたいです。

1) h5 ファイルの順次処理と dask 処理用のすべての cpu
2) h5 ファイルの並列処理をxコアで実行し、残り20-xを dask 処理に使用します。
3) リソースを分散し、h5 ファイル、各 h5 ファイル内の画像、および dask の残りのリソースを並列処理します。

助けてくれてありがとう！

基本的に行列方程式を解くプロジェクトに取り組んでいます

ここで、Aは 2000 年までにおよそ 10 000 000 の次元を持つ行列です (最終的にはこれを両方向に増やしたいと考えています)。

A明らかにメモリに収まらないため、これを並列化する必要があります。A.T.dot(A).dot(x) = A.T.dot(d)代わりに解決することでそれを行います。A.Tの次元は 2000 x 2000 になります。AとdをチャンクA_iに分割しd_i、 行に沿って と を計算A_i.T.dot(A_i)しA_i.T.dot(d_i)、これらを合計することで計算できます。並列化に最適です。私はこれを multiprocessing モジュールで実装できましたが、1) メモリの使用により、それ以上のスケーリングが難しく (A両方の次元で増加)、2) きれいではありません (したがって、保守が容易ではありません)。

Dask は、これらの問題の両方を解決するための非常に有望なライブラリのように思われ、いくつかの試みを行ってきました。私のA行列は計算が複雑です: 約 15 の異なる配列 (サイズは の行数に等しい) に基づいており、A関連するルジャンドル関数を評価するために反復アルゴリズムで使用されるものもあります。チャンクが小さい場合 (10000 行)、タスク グラフの作成に非常に長い時間がかかり、多くのメモリが必要になります (メモリの増加は反復アルゴリズムの呼び出しと一致します)。チャンクが大きい場合 (50000 行)、計算前のメモリ消費量ははるかに少なくなりますが、計算時に急速に使い果たされますA.T.dot(A)。で試してみましcache.Chestたが、計算が大幅に遅くなります。

タスク グラフは非常に大きく複雑でなければなりません - 呼び出しがA._visualize()クラッシュします。より単純なA行列では、これを直接行うことができます (@MRocklin による応答を参照)。単純化する方法はありますか？

これを回避する方法についてのアドバイスは大歓迎です。

おもちゃの例として、私は試しました

これも失敗し、アクティブなコアが 1 つだけですべてのメモリが使い果たされました。を使用chunks = (2e3, 1e5)すると、すべてのコアがほぼすぐに開始されますが、MemoryError1 秒以内に表示されます (現在のコンピューターには 15 GB あります)。chunks = (2e3, 1e4)より有望でしたが、すべてのメモリも消費してしまいました。

編集：寸法が間違っていたため、おもちゃの例のテストを打ち消し、残りの寸法を修正しました。@MRocklin が言うように、正しい寸法で動作します。私の問題にもっと関連していると思う質問を追加しました。

Edit2:これは、私がやろうとしていたことの非常に単純化された例です。問題は、 の列の定義に含まれる再帰にあると思いますA。

これは非常に複雑なタスク グラフにつながるようで、計算が始まる前に大量のメモリを占有します。

再帰をnumpy配列を使用して関数に配置することでこれを解決しましたが、多かれ少なかれ do A = x.map_blocks(...).

2 つ目の注意として、Aマトリックス タスク グラフを取得すると、A.T.dot(A)直接計算するとメモリの問題が発生するようです (メモリの使用量はあまり安定していません)。したがって、チャンクで明示的に計算し、結果を合計します。これらの回避策を使用しても、dask は速度と可読性に大きな違いをもたらします。

PythonでnetCDF4を使用して読んでいる非常に大きなnetCDFファイルがあります

このファイルのサイズ (1200 x 720 x 1440) は、ファイル全体を一度にメモリに入れるには大きすぎるため、一度にすべてを読み取ることはできません。最初の次元は時間を表し、次の 2 次元はそれぞれ緯度と経度を表します。

しかし、一度に 1 年ずつ読むのは非常に遅いです。以下のユースケースでこれを高速化するにはどうすればよいですか?

- 編集

チャンクサイズは 1

1. 年の範囲を読み取ることができます: nc_file.variables[変数名][0:100, :, :]

2. いくつかのユースケースがあります:

   年の年:

   /li>

dask DataFrame に変換して castra に保存しようとしている json ファイルのディレクトリがあります。それらの間に O(10**7) json レコードを含む 200 個のファイルがあります。コードは非常に単純で、主にチュートリアルの例に従っています。

32 コアのマシンで実行していますが、コードは 1 つのコアしか 100% 使用していません。ドキュメントからの私の理解は、このコードが並行して実行されるということです。なぜそうではないのですか？私は何かを誤解しましたか？

Dask の使用、特にその遅延計算と DAG 機能を検討しています。

未知の入力を持つ、適度に複雑な計算データがあります。事前にビルドしてから、さまざまな入力で使用できるようにしたいと考えています。

dict / tuple インターフェースでこれを行うことができると思います：

次にdsk、ポータブル ダグとして、x必要なものに置き換えることができます。(実際、最初は上記に含める必要はありませんでした)。

しかし、私たちはこれを行うことができdask.imperativeますか? 私の最初の結果は、私たちが到達できないことを示唆していますx:

方法はありますか？

Dask FAQ では、特異値分解 (SVD) について言及しています ( http://dask.pydata.org/en/latest/faq.html ) 。しかし、API に SVD が表示されません。PyPi の最新バージョンである 0.8.0 をインストールしました。

ありがとう、カール

Dask では、ジェネレーターがリストに変換されるのはいつですか?それとも、通常は遅延して消費されますか? たとえば、次のコードを使用します。

次の出力が得られます。ここでは、ジェネレーターが (合理的に) 検査のためにリストに変換されています。

合理的ではありますが、リストへの明示的な変換を必要とする、Python でのジェネレーターの通常の動作とは異なります。

foldでは、 ( ) を計算するとき、ジェネレーターのresult.compute()入力引数 ですか、それとも既にリストに変換されていますか?xadd_to_count

リストが非常に長い場合に興味があるため、遅延評価の方が効率的 b1 = db.from_sequence([10**6]*10)です。

で上記の問題を解決できるとbag.frequencies思いますが、遅延評価と効率的な削減についても同様の懸念があります。

私が理解していないDaskの基本的な側面はありますか、それとも単に怠惰なだけですか?これを自分で理解するためにコードをどこで調べることができますか?