問題タブ [spark-dataframe]

データフレームとしての条件で分割された寄木細工を読み取る方法、

これはうまくいきます、

day=1 to day=30のようなものを読み取ることができるか、(day = 5 to 6)またはのようなパーティションがあります。day=5,day=6

私が置く*と、30日すべてのデータが得られ、大きすぎます。

以前の RDD パラダイムでは、キーを指定して、各キーに対応する RDD 要素に操作をマップすることができました。1.5.1 の時点で、SparkR の DataFrame でこれを行う明確な方法がわかりません。私がやりたいことは、dplyr操作のようなものです:

現在、次の形式の大きな SparkR DataFrame があります。

idとで並べ替えましたtimestamp。

でグループ化しidたいのですが、集計はしたくありません。代わりに、各グループで一連の変換と計算を実行したいと考えています。たとえば、NA を埋めるために補間します (これはcollect、DataFramevalueを数値に変換したときに生成されます)。を使用してテストしましaggたが、計算は実行されているように見えますが、 で単一の値を返していないため、結果は返されませんmyfunc。

DataFrame を 1 つに落として実行すると、すべての操作がmyfunc正しく機能することに注意してください。実行にかかる時間 (タスクあたり約 50 秒) と例外がスローされないという事実に基づいて、すべてのs で実際に実行されていると思いますが、出力が必要です!filteridmyfuncid

任意の入力をいただければ幸いです。

Spark 1.3.1 を使用しています。

DataFrame データを Hive メタストアに保存/保存する方法は?

Hive で DataFrame を実行show tablesすると、Hive データベースにテーブルとして表示されません。にコピーhive-site.xmlしました$SPARK_HOME/confが、役に立ちませんでした (また、データフレームが Hive メタストアにも表示されません)。

Spark 1.4 バージョンを使用して、このドキュメントに従っています。

Hiveでスパークテーブルを分析するには?

DataFrame特定のパーティショニングに従って S3 に書き込む必要があります。コードは次のようになります。

データをかなり多数のpartitionByフォルダー (~400) に分割し、それぞれにほんの少しのデータ (~1GB) を入れます。ここで問題が発生します。デフォルト値spark.sql.shuffle.partitionsは 200 であるため、各フォルダー内の 1GB のデータは 200 個の小さな寄木細工ファイルに分割され、合計で約 80000 個の寄木細工ファイルが書き込まれます。これは多くの理由で最適ではなく、回避したいと考えています。

もちろん、spark.sql.shuffle.partitionsを 10 などのはるかに小さい数に設定することもできますが、この設定は、結合と集計のシャッフルのパーティション数も制御することを理解しているため、これを変更したくありません。

書き込まれるファイルの数を制御する別の方法があるかどうかは誰にもわかりませんか?

スパークのデータフレームの場合、次のように日時の差分を計算したいnumpy.diff(array)

編集:以前の質問の質について申し訳ありませんが、これがより明確になることを願っています:Sparkアプリケーションでは、次のJSONファイルのディレクトリ全体をロードしています:

後で使用するDataFrameために一時テーブルとして保存します。この Json では、「payload」ノードのフィールドは常に存在しますが、「masterdata」のサブノードはオプションです。次のステップは、次のように Json のサブノードごとに複数の DataFrame を作成することです。最初の部分を処理した後、Spark の状態は次のとおりです。

ここで問題が発生します。ディレクトリ内の JSON ファイルの 1 つに md2 ノードが含まれていない場合、NullPointException が原因で "md2" DataFrameshow()でも実行できません。collect()すべてのファイルに「md2」ノードがない場合は理解できるので、md2 DataFrame を作成できませんでしたが、この場合、md2 DataFrame には、ノード md2 を持たず、他のすべてを含む json ファイルからのデータがないことが予想されます。

技術的な詳細: ネストされたノードからデータを読み取るには、rdd.map と rdd.flatmap を使用し、それをDataFrameカスタム列名に変換します

ディレクトリ内のすべてのファイルにすべてのノードが含まれているときにアプリケーションを実行すると、すべてが機能しますが、単一のファイルが md2 ノードにない場合、アプリケーションは .show() または .collect() で失敗します

ところで、ノードが存在するが空の場合、すべて正常に動作します。

Spark でオプションの Json ノードをサポートしたり、rdd.map&flatmap 内の欠落したノードを処理したりする方法はありますか?

前の質問よりも明確であることを願っています

@Berylium リクエストで、md2 DataFrame を取得するために使用している rdd 操作を次に示します。

列の指数移動平均を計算する必要があるアプリケーションを Spark/Scala で作成しています。

私が直面している問題は、同じ列の以前に計算された値 (EMA_t-1) が必要なことです。mySQLを介して、MODELを使用するか、行ごとに更新できるEMA列を作成することでこれが可能になりますが、これを試してみましたが、Spark SQLまたはHiveコンテキストでは機能しません...アクセスできる方法はありますか?このEMA_t-1?

私のデータは次のようになります。

したがって、最初の値が最初の行の価格である新しい列を追加する必要があり、次に前の値を使用する必要があります: EMA_t = (price_t * 0.4) + (EMA_t-1 * 0.6) を計算するにはその列の次の行。私の EMA 列は次のようになります。

私は現在、Spark SQL と Hive を使用してそれを実行しようとしていますが、別の方法で実行できる場合は、これも大歓迎です! また、Spark Streaming を使用してこれを行う方法についても考えていました。私のデータはデータフレームにあり、Spark 1.4.1 を使用しています。

提供されたヘルプに感謝します！