問題タブ [spark-dataframe]

csv ファイルからデータフレームに読み込むサンプル アプリケーションがあります。メソッドを使用して、データフレームを寄木細工形式で Hive テーブルに格納できます df.saveAsTable(tablename,mode)。

上記のコードは問題なく動作しますが、作成日 (テーブルの列) に基づいてハイブ テーブルを動的に分割したい毎日のデータが非常に多いためです。

データフレームを動的に分割してハイブ ウェアハウスに格納する方法はありますか。を使用して挿入ステートメントをハードコーディングすることは控えたいhivesqlcontext.sql(insert into table partittioin by(date)....)。

質問は拡張機能と見なすことができます: DataFrame を Hive に直接保存する方法は?

どんな助けでも大歓迎です。

cassandrasql のデータフレームがあり、次のようなデータフレームのマップである列があります

完全なマップではなく、dfの列としてマップから特定のキーを選択するだけでなく、dfからいくつかの列を選択する必要があります

key1 -> value1 key2 -> value2 のマップがあるとしましょう ....

新しいデータフレームの列になるには、データフレームのマップから key1 のみを選択する必要があります。どうやってやるの

また、cassandrasqlcontext.sql を使用してデータフレームを取得しています。

私はApache Sparkの初心者です。

私の仕事は、2 つの CSV ファイルを読み取り、そこから特定の列を選択し、マージして集計し、結果を 1 つの CSV ファイルに書き込むことです。

例えば、

CSV1

CSV2

で 3 つ目の CSV ファイルを取得したい

両方の CSV をデータフレームにロードしています。そして、join,select,filter,dropデータフレームに存在するいくつかのメソッドを使用して3番目のデータフレームを取得できます

私はまた、いくつかを使用して同じことを行うことができますRDD.map()

hiveqlまた、次を使用して実行することで同じことを行うこともできますHiveContext

CSV ファイルが巨大な場合、どの方法が効率的かを知りたいのですが、その理由は何ですか?

データフレームから空のパーティションを削除する必要があります

2 つのデータフレームがあり、どちらも sqlContext を使用して作成されます。そして、データフレームは以下のように構築され、結合されます

これで、結合DFのパーティション数は200です。ここから、結合を使用する場合のデフォルトのパーティション数は200であることがわかります。

場合によっては、データフレーム/csv が大きくなく、多くの空のパーティションが取得され、コードの後半部分で問題が発生することがあります。

では、作成されたこれらの空のパーティションを削除するにはどうすればよいですか?

Apache Zeppelin ノートブックを使用して spark-csv [1] で CSV ファイルを Spark データ フレームにロードしようとしていますが、値のない数値フィールドをロードすると、その行のパーサーが失敗し、その行がスキップされます。

行がロードされ、データフレームの値が行をロードし、値が NULL に設定されて、集計が値を無視するようになると予想していました。

データ ファイルの内容は次のとおりです: /home/spark_user/data.csv

出力は次のとおりです。

zeppelin のログで、サンタの行を解析する際に次のエラーが表示されます。

だから、あなたは私にこれまでのところとても良いと言うかもしれません...そしてあなたは正しいでしょう;)

ここで、年齢などの追加の列を追加したいと思います。そのフィールドには常にデータがあります。

ここで、年齢に関するいくつかの統計を丁寧に尋ねます。

結果

すべて間違っています！サンタの身長がわからないため、行全体が失われ、年齢の計算はサムとキャスのみに基づいて行われますが、サンタの年齢は完全に有効です。

私の質問は、CSV をロードできるようにサンタの身長をプラグインするために必要な値は何かということです。スキーマをすべて StringType に設定しようとしましたが、その後

次の質問は、

API で、spark を使用して N/A 値を処理できることがわかりました。そのため、すべての列を StringType に設定してデータをロードし、クリーンアップを行ってから、以下に示すようにスキーマのみを適切に設定できるのではないかと考えました。

しかし、 df.na.replace() は例外をスローして停止します:

どんな助け、＆ヒントも大歓迎です!!

[1] https://github.com/databricks/spark-csv

HDFS システムから寄木細工のファイルを読み取りました。

この問題についていくつかのスレッドを読みました。しかし、それは私の場合ではありません。実際、寄木細工のファイルからテーブルを読み取っただけhead()ですcollect()。私の寄木細工のテーブルは次のようなものです：

私はspark-1.4.0-bin-hadoop2.6を使用しており、これをクラスターで実行します

ローカルでも試しましたが、同じ問題があります。

また、Scala でこの寄木細工のファイルを読み取ろうとしました。そして、collect() 操作を実行します。すべてがうまくいっているようです。したがって、SparkR に固有の問題である必要があります。

データフレームから選択した数の行をランダムに選択したいのですが、サンプルメソッドがこれを行うことは知っていますが、ランダム性が均一サンプリングであるべきだと心配していますか? それで、Dataframes での Spark のサンプル メソッドが統一されているかどうか疑問に思っていました。

ありがとう

テキストファイルの場合と同じように、xlsおよびxlsx（MS Excel）ファイルをsparkで行ごとに読み取りたいですか、それともどのようにしますか?

スパークを使用して、1 GB などの大きな xls ファイルを読み取るパフォーマンスを向上させたいと考えています。そのため、テキスト ファイルの場合と同様に、ファイルを部分的に読み取るためにスパークが必要です。

行ごとかどうかに関係なく、sparkでExcelファイルからデータを読み取るにはどうすればよいですか?

とにかくsparkを使用してxlsファイルのエントリを読みたいだけです。

提案してください。

ありがとう！！！

セットの分析をしようとしています。次のようなサンプル データ セットがあります。

注文.json

それは、ID を表す数字のリストである単一のフィールドだけです。

実行しようとしているSparkスクリプトは次のとおりです。

作成expandedしgroupedて問題ありません。一言で言えばexpanded、2 つの ID が同じ元のセットにあった、2 つの ID のすべての可能なセットのリストです。grouped自分自身と一致した ID を除外し、ID のすべての一意のペアをグループ化し、それぞれのカウントを生成します。のスキーマとデータ サンプルは次のgroupedとおりです。

だから、私の質問は、タプルのリストを得るために、各結果の最初の項目をグループ化するにはどうすればよいですか? 上記のサンプル データの場合、次のようなものが期待できます。

のスクリプトでわかるように、recs各行の最初のアイテムである「item1」で groupBy を実行することから始めると思いました。しかしその後は、この GroupedData オブジェクトに対するアクションが非常に制限されたままになります。実際には、sum、avg などの集計を行うだけです。各結果のタプルを一覧表示したいだけです。

この時点で RDD 関数を簡単に使用できましたが、それはデータフレームの使用から逸脱しています。データフレーム関数でこれを行う方法はありますか。

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

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