問題タブ [pyspark-sql]

すべて 4 コアと 16GB RAM を備えた 3 つの VM (つまり、1x マスター、2x スレーブ) で spark 1.6 を実行しています。

spark-master webUI に登録されているワーカーを確認できます。

Vertica データベースからデータを取得して作業したいと考えています。複雑なクエリを実行できなかったので、理解するためにダミーのクエリを試しました。ここでは簡単な作業と考えます。

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

@IPSLAVE出力は次のとおりです (注:スレーブ VM の IP:Portに置き換えます)。

ご覧のとおり、非常に長い時間がかかります。私のテーブルは実際には非常に大きい (約 2 億 2000 万行、それぞれ 11 フィールドを格納) ですが、このようなクエリは「通常の」SQL (pyodbc など) を使用してすぐに実行されます。

私はSparkを誤解/誤用していると思います.Sparkをよりうまく機能させるためのアイデアやアドバイスはありますか?

再開するには、関連付けられている同じグループの値に入れたい:

これが私が持っているものです:

そして私はこれが欲しい：

これらの 2 つのグループを作成するには、手動で行う場合の手順を次に示します。

• 行 1 : 1 は 2 に関連付けられているため、同じグループに属しています。これをグループ 1 と呼びましょう。
• 行 2 : 1 はグループ 1 にあり、現在 1 は 3 に関連付けられているため、3 もグループ 1 にあります
• 行 3 : 2 はグループ 1 にあり、3 もグループ 1 にあるため、グループ 1 に属します。
• 行 4: 4 はグループ 1 の値ではないため、2 という新しいグループを作成し、それを 5 に関連付けます。
• 行 5 : 5 にはグループ 2 があり、6 に関連付けられているため、グループ 2 があります。

これをSQLで解決するという考えはありますか。私がHiveまたはpysparkを使用していることを知っている

SparkSQL を使用して、論理 Cassandra 行 (CQL 行) ではなく、「ワイド行」を「反復」またはマップしたい状況があります。

基本的に、私のデータはtimestamp(パーティション キー) によって分割され、センサー ID であるクラスタリング キーがあります。

timestamp操作を実行したいそれぞれについて、簡単な例は do ですsensor1/sensor2。

データの局所性を維持することにより、SparkSQL で効率的にそれを行うにはどうすればよいでしょうか (そして、私のデータ モデルはこれらのタスクにかなり適していると思います)。

Cassandraコネクタで言及されているDatastaxに関するこの投稿を読みました。これを SparkSQL でどのように使用しますか?spanByspanByKey

疑似コードの例 (pySpark):

データ フレームがあり、データを 7 日間にロールアップし、一部の関数で集計を行いたいと考えています。

------のようなpyspark sqlデータフレームがあります

次のように、データフレームにウィンドウ関数を適用しました-

今、いくつかの集計を実行したい、つまり、次のようないくつかの Windows 関数を適用したい -

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

Hadoop でビルド済みの Apache Spark 1.6.0 を使用しています。

私はCassandraから読んでいます

そしてそれをデータフレームに変換したい：

スキーマは正しく推測されます。

しかし、データフレームを具体化すると、次のエラーが発生します。

stringに与えられたように聞こえpyspark.sql.types.TimestampTypeます。

これをさらにデバッグするにはどうすればよいですか?

pandasread_csv()関数をiso-8859-1使用して、次のようにファイルを読み取ります。

それでは、MLLib の word2vect を使用したいと思います。ただし、パラメーターとして RDD のみを受け入れます。そこで、次のように pandas データフレームを RDD に変換しようとしました。

とにかく、次の例外が発生しました。

エンコーディングパラメータのようなものがあるかどうかを確認するために、Pyspark のドキュメントにアクセスしましたが、何も見つかりませんでした。特定の pandas データフレーム列を Pyspark RDD に変換する方法について何か考えはありますか?

アップデート：

@zeros の回答から、これは、次のように columnn をデータフレームとして保存しようとしたものです。

それで：

そして、私は同じ例外を得ました: