問題タブ [pyspark-sql]

pyspark シェルを使用して糸クライアント モードで正常に動作する単純なハイブ クエリがあります。糸クラスター モードで実行すると、以下のエラーがスローされます。

クラスター情報: Hadoop 2.4、Spark 1.4.0-hadoop2.4、hive 0.13.1 スクリプトは、Hive テーブルから 10 列を取得し、いくつかの変換を行ってファイルに書き込みます。

完全なスタック トレース:

私はCSVを読んでいます

現在、PDF は Spark クラスター全体に分散されていますか、それともホスト マシンの環境に常駐していますか??

pyspark.sql DataFrame を作成するコードは次のとおりです。

そのため、sparkdf は次のようになります

ここで、numpy 配列 (またはリスト) を新しい列として追加したいと思います。

しかし、標準的な方法

失敗します。おそらく udf が最適ですが、DataFrame 行ごとに 1 つの異なる値を割り当てる、つまり new_col を反復処理する udf を作成する方法がわかりません。他の pyspark と pyspark.sql を見てきましたが、解決策が見つかりませんでした。また、scala ソリューションではなく、pyspark.sql 内にとどまる必要があります。ありがとう！

pyspark データフレームをグループ化し、特定の列の分散を計算したいと思います。平均の場合、これは非常に簡単で、次のように実行できます

ただし、差異については、関数サブモジュールに集計関数がないようです（これは非常に一般的な操作であるため、理由も疑問に思っています）

私はHiveでPyspark 1.2.1を使用しています。(アップグレードはすぐには行われません)。

私が抱えている問題は、Hive テーブルから選択してインデックスを追加すると、pyspark が long 値を int に変更するため、Long 型の列を持つが Integer 型の値を持つ一時テーブルになってしまうことです。(以下のコードを参照)。

私の質問は次のとおりです。(a) long を int に変更せずにインデックスのマージ (コードを参照) を実行するにはどうすればよいですか。または（b）問題を回避する他の方法でインデックスを追加します。または（c）結合する必要なくテーブルの列をランダム化しますか？

私が解決しようとしている根本的な問題は、ハイブ テーブル内の特定の列の順序をランダム化し、それを新しいテーブルに書き込みたいということです。これは、データが個人を特定できないようにするためです。元のテーブルとランダム化された列に増分インデックスを追加し、そのインデックスに結合することでそれを行っています。

テーブルは次のようになります。

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

より良い解決策がない場合、影響を受ける列を Python コードで long 型に強制します。これは...良くない。

コマンドの実行中に次のエラーが発生します。

Pythonでspark 1.2.0を使用しています。

私の問題は、フィールドの値がゼロの場合、SQLクエリで、それを他の値に置き換える必要があることです。

1.4.0 では機能するが 1.2.0 では機能しない case /coalese を試しました

ただし、1.2.0の場合、マップで同じことをしようとしました

これはエラーをスローします

spark 1.2.0 で目的を達成するために何ができるかわかりません。

助けてくれてありがとう、私はそれが今働いていると思います..列の順序が変更されたように見えることを除いて..しかし、それは問題ではないかもしれません。再度、感謝します

編集：

このアイデアは私を大いに助けてくれました。差し迫った問題を解決するには少し修正が必要でした。

「for」での構文エラーを回避するために、元のソリューションを変更しました。

メソッドの呼び出しは次のとおりです。

ただし、今 iifdatadf にアクセスしようとしていますが、

次のエラーが発生しています

助けていただければ幸いです。

pyspark 1.4.1 でSpark 1.4ウィンドウ関数を使用しようとしています

しかし、ほとんどの場合、エラーまたは予期しない結果が得られます。これはうまくいくはずだと思う非常に簡単な例です：

ご覧のとおり、rowsBetweenフレーム指定を追加すると、ウィンドウ関数rank()もlag/lead()それを認識しません:「ウィンドウ関数はフレーム指定を取りません」。

rowsBetweenフレームの仕様を少なくとも省略した場合、lag/lead()例外はスローされませんが、(私にとっては) 予期しない結果が返されます: always None。そして、rank()まだ別の例外で機能しません。

ウィンドウ関数を正しく取得するのを手伝ってくれる人はいますか?

アップデート

よし、これは pyspark のバグのように見え始めます。純粋な Spark (Scala、spark-shell) で同じテストを用意しました。

rowsBetweenScala では を適用できませんが、 と を省略した場合は両方とも期待どおりに動作しrank()ます。lag()/lead()rowsBetween

Spark SQL の助けを借りて、特定のグループ カテゴリに属する​​すべてのビジネス アイテムを除外しようとしています。

データは JSON ファイルからロードされます。

ファイルのスキーマは次のとおりです。

レストラン ビジネスに関連するすべてのビジネスを抽出しようとしています。

しかし、予想される列のタイプは文字列である必要があることを理解しているため、機能しませんが、私の場合、これは配列です。それについて私に例外を教えてくれます：

私が欲しいものを手に入れるための他の方法を提案してもらえますか？

次のコードを使用して、給与がしきい値を超えている人の平均年齢を取得しています。

列 age は数値 (float) ですが、それでもこのエラーが発生します。

groupBy関数や SQL クエリを使用せずに平均などを取得する他の方法を知っていますか。