apache-spark

6熱

1答えて

Scalaコンパイラは未使用の変数値をどのように処理しますか？スカラとスパークを使用

、私は、次の構造を有する：PairRDDに val rdd1: RDD[String] = ... val rdd2: RDD[(String, Any)] = ... val rdd1pairs = rdd1.map(s => (s, s)) val result = rdd2.join(rdd1pairs) .map { case (_: String, (e: Any

9熱

3答えて

SparkとSparkSQL：ウィンドウ関数を模倣する方法は？

説明データフレームdf id | date --------------- 1 | 2015-09-01 2 | 2015-09-01 1 | 2015-09-03 1 | 2015-09-04 2 | 2015-09-04 私はランニングカウンタまたはインデックスを作成したいを考える、同じIDでグループ化されたし、そのグループに日付順に並べ替えられ、したがって、 id

7熱

3答えて

Mavenを使ったScala Sparkプロジェクトのパッケージ化と実行

私はScalaでSparkを使用するアプリケーションを作成しています。私はMavenを使用してアプリケーションをパッケージ化し、"uber" or "fat" jarを構築するときに問題にぶつかっています。私が直面している問題は、IDEの内部でアプリケーションを正常に動作させること、またはJavaクラスパスとして依存性の非uber-jarバージョンを提供する場合ですが、 java -Xmx2G

6熱

4答えて

RDDまたはデータフレームがキャッシュされているかどうかを確認するにはどうすればよいですか？

データフレームdf1を作成しました。私はこれをdf1.cache（）を使ってキャッシュしました。これがキャッシュされているかどうかをどのように確認できますか？キャッシュされたRDDやデータフレームをすべて見ることができる方法もあります。

8熱

2答えて

スパークデータフレーム内の複数の行を単一の行にマージする

私は2つの列を持つデータフレームを持っています：タイムスタンプ、値タイムスタンプはエポックと値がフロート値なので時間です。行を平均値に分単位でマージしたい。これは、タイムスタンプが同じラウンド・イン（エポックから60秒間隔）のすべての行を取得し、それらを単一の行にマージすることを意味します。値の列はすべての値の平均になります。第4行は同じ分（1441637160% 60 == 0、14416

24熱

1答えて

スパークストリーミング：なぜ数MBのユーザー状態を処理するために内部処理コストが高いのですか？

私たちの実験によると、ステートフルなSpark Streamingの内部処理コストは、状態が100万を超えるオブジェクトになるとかなりの時間がかかります。その結果、不安定な動作（処理時間>バッチ間隔）を避けるために、バッチ間隔を増やす必要があるため、遅延が発生します。以下のコードで再現できるので、私たちのアプリの詳細とは関係ありません。 Spark内部処理/インフラストラクチャのコストは、ユーザ

8熱

1答えて

Pythonクラスを使ってRDDを処理する方法は？

私は、Pythonのクラスとしてスパークでモデルを実装していて、いつでも私はそれが失敗したRDDにクラスメソッドをマッピングしてみてください。私の実際のコードはもっと複雑ですが、この簡易版は、問題の核心に取得します。私は、（例えば）ので、のようなモデルを実行した場合、 class model(object): def __init__(self): self.data = s

6熱

2答えて

RDDからRDDのコレクションを作成する方法は？

私はRDD[String]、wordRDDを持っています。また、文字列/単語からRDD [String]を作成する関数もあります。私はwordRDDの各文字列のための新しいRDD を作成したいと思います。ここに私の試みです： 1）スパークは、ネストされたRDDSサポートしていないため失敗しました： var newRDD = wordRDD.map(word => { // execute

6熱

2答えて

RDDでSpark/ScalaコンパイラがtoDFを見つけられないのはなぜですか[Map [Int、Int]]？

なぜ次のエラーが発生するのですか？ scala> import sqlContext.implicits._ import sqlContext.implicits._ scala> val rdd = sc.parallelize(1 to 10).map(x => (Map(x -> 0), 0)) rdd: org.apache.spark.rdd.RDD[(scala.collec

8熱

1答えて

アプリケーションに与えられるSpark Clusterの量を制御するものは何ですか？

スタティックパーティショニングのドキュメントhttps://spark.apache.org/docs/latest/job-scheduling.htmlのこのページでは、「この方法では、各アプリケーションに最大限のリソースを割り当てることができます」と記載されています。私はこれらの最大のリソースは何ですか？私はエグゼキュータ設定ごとにメモリを見つけました（動的パーティション化ではちょうど下に