問題タブ [dask]

を使用して 1.25 GB のデータセットを Python にインポートしようとしていますdask.array

ファイルは 1312*2500*196uint16の配列です。float32後で処理するために、これを配列に変換する必要があります。

この Dask 配列を でつなぎ合わせることができましたがuint16、 に変換しようとするとfloat32、メモリ エラーが発生します。

チャンクサイズに何をしても問題ありません。常にメモリエラーが発生します。

配列を 100 行で連結して配列を作成します (2500 次元を 100 行の小さな断片に分割します。これは、ファイルを読み取ってから配列を作成するために使用する必要がある画像ファイルをdaskネイティブに読み取ることができないためです。以下では、 「できるだけ短い」コード スニペットを提供します。.RAWnumpy.memmap()

私は2つの方法を試しました：

1）完全なuint16配列を作成してから、次のように変換してみてくださいfloat32：

(注:memmapは 1312x100x196 の配列で、行の範囲は 0 から 24 です)

そして、私は使用します

方法 2:

どちらの方法でもメモリ エラーが発生します。

これには何か理由がありますか？

dask配列は最大 100 GB のデータセット計算を実行できると読みました。

すべてのチャンク サイズを試しました (10x10x10 から 1 行まで)

メモリより大きいデータセットの問題を解決しています。元のデータセットは .csv ファイルです。列の 1 つは、musicbrainz サービスからのトラック ID 用です。

私がすでにしたこと

.csv ファイルをdaskで読み取り、パフォーマンスを向上させるためにディスク上でcastra形式に変換しました。また、musicbrainz API にクエリを実行し、 peewee を使用して sqlite DB にいくつかの関連する結果を入力しました。プロセスに数日かかり、障害が発生した場合にデータを失いたくなかったため、別の dask.dataframe の代わりに DB を使用することにしました。

私はまだデータを実際に分析し始めていませんでした。データの再配置中になんとか十分な混乱を引き起こしました。

現在の問題

SQL DB から dask / castra データフレームに列を結合するのに苦労しています。実際、これが実行可能かどうかはわかりません。

代替アプローチ

タスクに最適なツールを選択する際にいくつかの間違いを犯したようです. カストラはおそらく十分に成熟しておらず、それが問題の一部だと思います. さらに、pandas と peewee's not で使用されているため、peewee を優先して SQLAlchemy を選択する方が良い場合があります。

Blaze + HDF5 は、dask + castra の優れた代替手段として機能する可能性があります。これは主に、HDF5 が castra よりも安定性/成熟度/完全性が高く、blaze がデータ ストレージに関する意見が少ないためです。たとえば、SQL DB のメイン データセットへの結合を簡素化できます。

一方、私は pandas に精通しており、dask は「同じ」API を公開しています。dasを使用すると、並列処理も得られます。

TL;DR

メインのデータセットに参加する必要がある、メモリ データセット + sqlite DB よりも大きなデータセットがあります。dask + castra (dask.dataframe に関連する他のデータ ストアがわからない) を使用し、SQLAlchemy を使用して SQL DB の一部を一度に pandas のデータフレームにロードするかどうかは疑問です。私が考える最良の代替案は、代わりにブレイズ + HDF5 に切り替えることです。この場合、あなたは何を提案しますか？

その他のオプション/意見は大歓迎です。これがSOにとって十分に具体的であることを願っています。

で動作するコードがありますがpandas、使用するように変換するのに問題がありますdask。ここに部分的な解決策がありますが、割り当て先の列の名前として変数を使用することはできません。

作業pandasコードは次のとおりです。

壊れたdaskコードは次のとおりです。

cこれにより、(必要なもの)の値を変更するのではなく、呼び出された新しい列に結果が割り当てられますdata[c]。列名を変数にすることができれば、新しい列を作成しても問題ありません。たとえば、これが機能した場合：

明らかな理由により、python は式を an の左側に置くことを許可しないため、=の前の値を無視しますname。

列の名前に変数を使用するにはどうすればよいですか? for ループは、コピー/貼り付けするよりもはるかに多く繰り返されます。

で動作するコードがありますがpandas、使用するように変換するのに問題がありますdask。ここに部分的な解決策がありますが、作成/割り当て先の列の名前として変数を使用することはできません。

作業pandasコードは次のとおりです。

dask私が望むことをしないコードは次のとおりです。

cこれにより、(必要なもの)の値を変更するのではなく、呼び出された新しい列に結果が割り当てられますdata[c]。列名を変数にすることができれば、新しい列を作成しても問題ありません。たとえば、これが機能した場合：

明らかな理由から、python は式を an の左側に置くことを許可せず、=の前の値を無視しますname。

割り当て先の列の名前に変数を使用するにはどうすればよいですか? ループは、コピー/貼り付けするよりもはるかに多く繰り返されます。

このようなバッチモードでdasがうまく機能することを理解しています

1. チャンクの数が不明または無限でさえあるストリーミング channel を処理するために dask を使用できますか?
2. インクリメンタルな方法で計算を実行できますか。たとえば、上記の「分析」ステップで進行中のチャンクを処理できますか?
3. すべてのデータ チャンクが判明した後でのみ「get」操作を呼び出す必要があります。「get」が呼び出された後に新しいチャンクを追加できますか

次のことを行う python スクリプトがあります。データの入力ファイル (通常はネストされた JSON 形式) を受け取ります。ｉｉｉ．データを所望のフォーマットに操作する別の関数に１行ずつデータを渡す。最後に、出力をファイルに書き込みます。

これがこれを行う私の現在の単純なpython行です...

これは機能しますが、python GIL がサーバー上の 1 つのコアに制限しているため、特に大量のデータでは、非常に遅くなります。

私が通常扱うデータの量は、gzip で圧縮された約 4 ギガですが、時には数百ギガの gzip で圧縮されたデータを処理する必要があります。必ずしもビッグデータではありませんが、すべてをメモリ内で処理することはできず、Python の GIL では処理が非常に遅くなります。

データ処理を最適化するソリューションを探しているときに、das に出会いました。当時の私には PySpark が当然の解決策のように思えましたが、dask の約束とそのシンプルさに心を奪われ、試してみることにしました。

dask とその使用方法について多くの調査を行った後、現在のプロセスを再現するための非常に小さなスクリプトを作成しました。スクリプトは次のようになります。

これは機能し、元の非 Dask スクリプトと同じ結果を生成しますが、サーバー上で1 つの CPU しか使用しません。だから、それはまったく役に立ちませんでした。実際、それはより遅いです。

私は何を間違っていますか？何か不足していますか？私はまだdaskにかなり慣れていないので、何かを見落としているか、まったく別のことをすべきかどうかを教えてください.

また、必要なことにサーバーの全容量 (つまり、すべての CPU) を使用するための dask の代替手段はありますか?

ありがとう、

T