dd.from_bcolz() が呼び出されるとすぐに処理を開始するのはなぜですか (N 列が増え、文字列型の列があると大きくなります)。

そして、 dd.read_hdf() は、呼び出されたときに多くの処理を行いませんが、dask.dataframe が使用された場合にのみ、read_hdf() は HDF5 チャンクを読み取り、チャンクごとに処理します...

hdf5テーブルが最大1200列を超えることができず、データフレームが配列の列をサポートしていないという唯一の問題です。結局、hdf5形式は列ベースではありません...

そして、N (nrows) が成長すると、さらに悪化します。

私は現在配布されているpythonパッケージを使用していますが、Python APIでセットアップしたときにWeb UIのアドレスを取得する方法を知っている人はいますか?

エグゼキュータ = エグゼキュータ()

または、常にコマンドラインで設定する必要があります

タスクスケジューラ

Web UI のアドレスを取得するには?

パンダを指定すると、次のdf方法で何かを行う前にコピーできます。

daskデータフレームオブジェクトでこれを行うにはどうすればよいですか?

TLDR :ダスクバッグからダスクデータフレームを作成しました。dask データフレームは、すべての観測 (イベント) を列として扱います。したがって、各イベントのデータの行ではなく、各イベントの列があります。目標は、パンダが df.T を使用してデータフレームを転置できるのと同じ方法で、列を行に転置することです。

詳細:私のタイムラインのサンプル Twitter データはこちらにあります。私の出発点に到達するために、これは json をディスクから に読み取り、dask.bagそれを に変換するコードですdask.dataframe

問題私の個々のイベント (つまりツイート) はすべて、列と行として記録されます。原則tidyに従って、イベントごとに行を作成したいと思います。 pandasにはデータフレームの転置メソッドがあり、dask.array には配列の転置メソッドがあります。私の目標は、同じ転置操作を行うことですが、dask データフレームで行います。どうすればいいですか？

1. 行を列に変換する

ソリューションの編集

このコードは元の転置の問題を解決し、保持する列を定義して残りを削除することで Twitter json ファイルをクリーンアップし、関数をシリーズに適用して新しい列を作成します。次に、はるかに小さいクリーンなファイルをディスクに書き込みます。

dask を使用して、あるタスクから別のタスクに取得できる変数を持つ方法はありますか。つまり、ワーカーでロックして、別のタスクを実行するときに同じワーカーで取得できる変数です。

この方法でdaskを使用してcsvファイルを読みました：

次に、行ごとに単純なロジックを適用したいと思います。これは、パンダではかなりうまく機能します。

残念ながら、コードの最後の行で次のエラーが生成されます:値の長さがインデックスの長さと一致しません

私は何を間違っていますか？

次のコードを使用して、Pandas データフレームを列ごとに離散化できます。

パラメータとして loc (列の中央値) と scale (列の中央絶対偏差) を使用して、各列を離散化したいと考えています。

小さなデータフレームでは、必要な時間は許容されます (シングル スレッド ソリューションであるため)。

ただし、より大きなデータフレームでは、計算を高速化するために、より多くのスレッド (またはプロセス) を活用したいと考えています。

私はこの問題の解決策を提供するはずのDaskの専門家ではありません。

ただし、私の場合、離散化はコードで実行可能である必要があります。

ただし、ここでの問題は、locとscaleが列の値に依存しているため、適用前または適用中に各列に対して計算する必要があることです。

どうすればそれができますか？

現在、データ分析に Pandas と Spark を使用しています。Dask が並列化された NumPy 配列と Pandas DataFrame を提供していることがわかりました。

Pandas は、Python でデータ分析を行うための簡単で直感的なツールです。しかし、システムメモリが限られているため、Pandas で複数の大きなデータフレームを処理するのは困難です。

簡単な答え:

Apache Spark は、分散コンピューティング、SQL クエリ、機械学習などを組み合わせた包括的なフレームワークであり、JVM 上で実行され、Hadoop などの他のビッグ データ フレームワークと一緒にデプロイされるのが一般的です。... 一般的に Dask は Spark よりも小型で軽量です。

http://dask.pydata.org/en/latest/spark.htmlから以下の詳細を知ることができます

• Daskは軽量です
• Dask は通常、単一のマシンで使用されますが、分散クラスターでも適切に動作します。
• Dask to は、並列配列、データフレーム、機械学習、およびカスタム アルゴリズムを提供します。
• Daskはそれ自体がPythonライブラリであるため、Pythonユーザーにとって利点があり、問題が発生した場合のシリアル化とデバッグがよりスムーズに行われます.
• Dask は、ユーザーがより複雑な並列アルゴリズムを表現できるように、高度な理解を放棄します。
• Dask は軽量で、既存のコードやハードウェアに簡単に統合できます。
• すべてを行う単一のプロジェクトが必要で、すでにビッグデータ ハードウェアを使用している場合は、Spark が安全な選択です
• Spark は通常、小規模から中規模のクラスターで使用されますが、単一のマシンでも適切に動作します。

以下のリンクから Dask についてより多くのことを理解しています https://www.continuum.io/blog/developer-blog/high-performance-hadoop-anaconda-and-dask-your-cluster

• Python で Pandas、NumPy、またはその他の計算を使用しているときに、1 台のマシンでメモリの問題、ストレージの制限、または CPU の境界が発生している場合、Dask は 1 台のマシンのすべてのコアでスケールアップまたはスケールアウトするのに役立ちます。クラスタ全体のすべてのコアとメモリで。
• Dask は単一のマシンでうまく機能し、ラップトップのすべてのコアを利用して、メモリよりも大きなデータを処理します
• 数百のノードを持つクラスターで、回復力と弾力性を持ってスケールアップします。
• Dask は、Hadoop Distributed File System (HDFS) や Amazon S3 など、さまざまな形式やストレージ システムのデータを Python からネイティブに処理します。Anaconda と Dask は、Cloudera CDH や Hortonworks HDP など、既存のエンタープライズ Hadoop ディストリビューションと連携できます。

http://dask.pydata.org/en/latest/dataframe-overview.html

制限事項

Dask.DataFrame は Pandas インターフェイス全体を実装していません。これを期待しているユーザーはがっかりするでしょう。特に、dask.dataframe には次の制限があります。

1. ソートされていない列から新しいインデックスを設定するとコストがかかります
2. groupby-apply や並べ替えられていない列の結合などの多くの操作では、インデックスを設定する必要があり、前述のようにコストがかかります。
3. Pandas API は非常に大きいです。Dask.dataframe は、多くの pandas 機能や NDFrame などのより風変わりなデータ構造を実装しようとはしません。

Dask 開発者に感謝します。とても有望な技術のようです。

全体として、Dask は Spark よりも使いやすいことがわかります。Dask は Pandas と同じくらい柔軟で、より多くの CPU で並列に計算する能力が高くなります。

Dask に関する上記の事実をすべて理解しています。

では、Dask で処理できるデータ量 (テラバイト単位) はおよそどれくらいでしょうか?

10 個のワーカーと合計 40 個のコアを持つダッシュ グリッドがあるとします。これは共有グリッドなので、自分の作業で完全に飽和させたくありません。実行するタスクが 1000 個あり、一度に最大 20 個のタスクを送信 (およびアクティブに実行) したいと考えています。

具体的に言うと、

キューのシステムをセットアップすると

これは機能しますが、すべてのタスクがグリッドにダンプされ、飽和状態になります。理想的には、次のことができます。

ここのドキュメントの例では、maxsizeキューを使用できるようです。しかし、ユーザーの観点から見ると、私はまだバックプレッシャーに対処しなければならないようです。理想的 daskには、これを自動的に処理します。