pyspark

1熱

1答えて

私はディレクトリにn個のzipを持っていますが、それらのそれぞれを抽出し、zip内にあるファイルまたは2つのデータからいくつかのデータを抽出してグラフDBに追加したいとします。私はこの全部のために逐次的なpythonスクリプトを作ったが、私はそれをsparkのために変換することに固執している。私のzipはすべてHDFSディレクトリにあります。そして、グラフDBはNeo4jです。私はまだneo4jと

3熱

1答えて

read_csv（）のS3リーダーは、ファイルを最初にディスクにダウンロードするのですか、それともストリーミングを使用していますか？

私は、S3からデータを読み取るためにパンダでread_csvを使用しています。最初にファイルをローカルディスクにダウンロードしてからメモリにロードするか、または着信データをメモリに直接ストリームするかどうか、ディスクにロードする途中のステップは必要ありません。 SparkのsqlContext.read.load機能について同じ質問があります。

1熱

1答えて

hiveContext経由のSpark JobでHive関数を使用する

私はHive 1.2とSpark 1.4.1を使用しています。 hive> select row_number() over (partition by one.id order by two.id) as sk, two.id, two.name, one.name, current_date() from avant_source.one one inner join avant_sou

0熱

1答えて

タプルのRDDでpysparkデータフレーム列を更新する

私はユーザー情報が文字列であるデータを扱っています。これらの文字列に固有の整数値を割り当てたいと思います。私はこのスタックオーバーフローポストhereにいくらか従っていました。私は最終的にALSモデルを実行されて何をしたいのか data = data.map(lambda x: Rating(int(user.lookup(x[0])), int(x[1]), float(x[2]))) を

1熱

1答えて

スパーク環境（IBM Bluemix）のspark-csvまたはpyspark-csv

Spark on Bluemixに多数の大きなCSVファイルをロードする必要があります。私はsc.testFileでそれを行い、それをマップすることができますが、それは反復的で面倒なコードを必要とします。 databricks spark-csvパッケージ、またはpyspark-csvを環境に追加/ロードする方法はありますか（試しましたが、気に入らなかったのですか）。私はパンダを使ってその例を

11熱

4答えて

Sparkの--py-filesが動作しないようです。

SparkでPythonを使用する際に問題があります。私のアプリケーションには、numpy、pandas、astropyなどのいくつかの依存関係があります。クラスタ上のノードにはHDFS以外の共通のマウントポイントやファイルシステムがないため、virtualenvを使用してすべての依存関係の環境を作成することはできません。したがって、私はspark-submit --py-filesを使用して立ち

2熱

2答えて

ハイブコンテキストがpysparkで一時テーブルを認識しない - AnalysisException：「テーブルが見つからない」

私はローカルモードで実行されているpyspark（1.6.1）を使用しています。私はCSVファイルからのデータフレームを持っており、dense_rank（）カラムを追加する必要があります。 sqlContextはウィンドウ関数をサポートしていませんが、HiveContextはサポートしています。 hiveContext = HiveContext(sc) df.registerTempTabl

-1熱

1答えて

IndexedRowMatrixのpysparkラッパーmultiply（）

IndexedRowMatrixクラスのPySparkラッパーには、すべてのメソッドが含まれていません。具体的には、ラップするJava実装に含まれていてもmultiply（）メソッドがありません。しかし def multiply(self, other): other_java_matrix = other._java_matrix_wrapper._java_model ja

-1熱

1答えて

スパークでネストされたジョブを実行中

私はPySparkを使用しています。私はs3にgziped jsonファイルのリストを持っています。これは、アクセスして、変換して、パーケットでs3にエクスポートする必要があります。各jsonファイルには約100k行が含まれていますので、パラレル化するとそれほど意味がありません（しかし、並列化することはできます）が、並列化したファイルは約5kです。私のアプローチは、スクリプトにjsonファイルリス

1熱

1答えて

Pyspark - ラムダ内で関数を呼び出すとインポートエラーが発生する

私はかなり簡単なことをしようとしています。私はデータフレームの一部としてdatetimeオブジェクトを持っています。地図を作成するときに、日付を特定の方法でフォーマットしたいと思います。提出されたときにが unique = df.map(lambda x: (x.id,[[format_date(x.t),x.val]]))\ .reduceByKey(lambda x,y: x+y)