問題タブ [apache-spark-sql]

Spark SQL と Spark Streaming を統合するときの Not Serializable 例外に加えて

私のソースコード

JavaSQLContext も ForeachRDD ループの外で宣言されていますが、まだ NonSerializableException が発生しています

14/12/23 23:49:38 エラー JobScheduler: ジョブ ストリーミング ジョブの実行中にエラーが発生しました 1419378578000 ms.1 org.apache.spark.SparkException: org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala でタスクをシリアル化できません:166) org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:158) で org.apache.spark.SparkContext.clean(SparkContext.scala:1435) で org.apache.spark.rdd.RDD .map(RDD.scala:27​​1) at org.apache.spark.api.java.JavaRDDLike$class.map(JavaRDDLike.scala:78) at org.apache.spark.sql.api.java.JavaSchemaRDD.map(JavaSchemaRDD) .scala:42) で com.basic.spark.NumberCount$2.call(NumberCount.java:79) で com.basic.spark.NumberCount$2.call(NumberCount.java:67) で org.apache.spark.streaming. api.java.JavaDStreamLike$$anonfun$foreachRDD$1.apply(JavaDStreamLike.scala:27​​4) at org.apache.spark.streaming.api.java.JavaDStreamLike$$anonfun$foreachRDD$1.apply(JavaDStreamLike.scala:27​​4) at org.apache.spark.streaming.dstream.DStream$ $anonfun$foreachRDD$1.apply(DStream.scala:529) at org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1.apply(DStream.scala:529) at org.apache.spark.streaming. dstream.ForEachDStream$$anonfun$1.apply$mcV$sp(ForEachDStream.scala:42) at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:40) at org.apache. spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:40) で scala.util.Try$.apply(Try.scala:161) org.apache.spark.streaming.scheduler.Job.run で(Job.scala:32) org.apache.spark.streaming.scheduler で。JobScheduler$JobHandler.run(JobScheduler.scala:171) で java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) で java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) で java .lang.Thread.run(Thread.java:724) 原因: java.io.NotSerializableException: org.apache.spark.sql.api.java.JavaSQLContext at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1181) java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1541) で java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1506) で java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1429) で java.io. java.io.ObjectOutputStream の ObjectOutputStream.writeObject0(ObjectOutputStream.java:1175)。defaultWriteFields(ObjectOutputStream.java:1541) で java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1506) で java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1429) で java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java) :1175) で java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1541) で java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1506) で java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1429) で Java .io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1175) で java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:347) org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:42) で org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:73) at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:164) ... 20 以上

何か提案があればよろしくお願いします。

私は 3 つの SchemaRDD (HDFS の Parquet ファイルに格納された、それぞれが 100 万件のレコードのオーダー) を使用して 3 方向の結合を行っています。

スキーマは次のとおりです。

• table1 には、id、group_id、t2_id、および日付の 4 つのフィールドがあります。
• table2 には、id、group_id、および t3_id の 3 つのフィールドがあります。
• table3 には、id、group_id、date の 3 つのフィールドがあります。

グループ内の table1 と table3 の関係を把握しようとしています。

私が使用する SQL クエリは次のようになります。

しかし、私はSparkでそれをやろうとしています:

したがって、これは機能しているように見えますが、同じ (3 ユニット、EMR) クラスターで impala よりも実行速度が大幅に遅くなります。これは正しい方法ですか？これをより効率的にする方法はありますか？

助けてくれてありがとう

以下の私のコードのアルゴリズム
Step1。1 つの hbase エンティティ データを hBaseRDD に取得する

Step2 . hBaseRDD を rowPairRDD に変換する

Step3 . rowPairRDD を schemaRDD に変換する

Step4 . spark sql を使用して、最初の単純な sql クエリを実行します。

Step5 . spark sql を使用して、2 番目の単純な sql クエリを実行します。

Step6. spark sql を使用して、3 番目の単純な sql クエリを実行します。

テスト結果は次のとおりです。

テスト ケース 1 :

300,000 レコードを挿入すると、hbase エンティティがコードを実行します。

• 最初のクエリには 60407 ミリ秒が必要です
• 2 番目のクエリには 838 ミリ秒が必要です
• 3 番目のクエリには 792 ミリ秒が必要です

hbase Api を使用して同様のクエリを実行すると、2000 ミリ秒しかかかりません。どうやら、最後の 2 つの spark sql クエリは、hbase api クエリよりもはるかに高速です。
最初の spark sql クエリは、hbase からデータをロードするのに多くの時間を費やしていると思います。
したがって、最初のクエリは最後の 2 つのクエリよりもはるかに遅くなります。効果は期待できると思います

テスト ケース 2 :

400,000 レコードを挿入すると。hbase エンティティ、コードを実行します。

• 最初のクエリには 87213 ミリ秒が必要です
• 2 番目のクエリには 83238 ミリ秒が必要です
• 3 番目のクエリには 82092 ミリ秒が必要です

hbase Api を使用して同様のクエリを実行すると、3500 ミリ秒しかかかりません。どうやら 3 つの spark sql クエリは、hbase api クエリよりもはるかに遅いようです。
また、最後の 2 つの Spark SQL クエリも非常に遅く、パフォーマンスは最初のクエリと似ています。パフォーマンスを調整するにはどうすればよいですか?

PostgreSQL に 100 万行と 100 列以上のデータ ソースがあり、Spark SQL を使用したいので、このデータ ソースを変換してSchemaRDD.

Spark SQL プログラミング ガイドでは2 つのアプローチが紹介されています。1 つはリフレクションによるものです。

100以上の列があるので、これは面倒です。

もう 1 つのアプローチは、「プログラムによるスキーマの指定」です。つまり、次のように定義する必要があります。

これは私にとっても退屈です。

PostgreSQL実際には、クラスを使用してデータベースをロードするため、まだ別の問題がありますが、コンストラクターのパラメーターでJdbcRDDスキーマを定義する必要があることもわかりました。次のようになります。mapRowJdbcRDD

この API は、自分でスキーマを作成するように要求します。さらに悪いことに、これJdbcRDDをに変換するために同様のことをやり直す必要があります。これSchemaRDDは非常に扱いにくいコードです。

では、このタスクに最適なアプローチは何かを知りたいですか?

タプルごとに一意の ID を生成する必要がある [ETL ディメンション処理での代理キーの生成に相当]

私は40億行を扱います。生成された代理キーの最大値を取得する 3 行目のパフォーマンスが少し心配です。

同じことを達成するためのより良いクリーンな方法はありますか?

Spark SQLクエリに使用されるdataRDDとnewPairDataRDDの2つのspark RDDがあります。アプリケーションが初期化されると、dataRDD が初期化されます。指定された 1 つの hbase エンティティ内のすべてのデータが dataRDD に格納されます。

クライアントの sql クエリが来ると、私の APP はすべての新しい更新と newPairDataRDD への挿入を取得します。dataRDD は newPairDataRDD を結合し、spark SQL コンテキストでテーブルとして登録します。

dataRDD に 0 レコード、newPairDataRDD に新たに挿入された 1 レコードも見つかりました。結合には 4 秒かかります。それは遅すぎる

合理的ではないと思います。速くする方法を知っている人はいますか？以下のような簡単なコードに感謝します

Spark Web UI から、以下を確認できます。どうやら結合には4が必要なようです

完了したステージ (8)

ステージ ID 説明 提出された期間 タスク: 成功/合計入力 シャッフル読み取り シャッフル書き込み

6 SparkPlan.scala:85+details で収集 2015 年 1 月 4 日 8:17 2 秒 8 月 8 日 156.0 B

SparkSqlQueryForMarsNew.java:389+details での 7 ユニオン 2015 年 1 月 4 日 8:17 4 秒 8 月 8 日 64.0 B 156.0 B

次のようなテーブルがあるとしましょう：

およびインデックス:

SQL コンテキストを使用して本を照会したい。私がやっていることは次のとおりです。

次のクエリはCassandraでは機能しません。

「book1」のみを持つユーザーを返します。同様のクエリを試しましbooks CONTAINS ('book1', 'book2')たが、どれも機能しませんでした。

登録されたテーブルでできることは次のとおりです。

私がやりたいのは、次のような本によるクエリです。

または同様のクエリ。

しかし、うまくいきません。0 レコードを返します。asという名前のインデックステーブルを登録しようとしましuser_books_idxたが、うまくいきませんでした。インデックス付きコレクションに対してクエリを実行できますか? どうすればできますか？

私は Scala と Apache Spark の初心者で、Spark SQL を使用しようとしています。リポジトリのクローンを作成した後bin/spark-shell、次のように入力して実行して Spark シェルを起動しました。

すべてが期待どおりに機能しました。users.txtファイルは次のようなものです。

その後、スタンドアロン プロジェクトを作成しようとし、sbt. にリストされている依存関係build.sbtは次のとおりです。

同じ手順を実行すると、次の行でクラッシュします。

このエラーで：

何が問題ですか？

アップデート：

わかりました、問題はSpark SQLではないと思いますが、実行することさえできないので、Spark自体ですusers.collect(). 代わりに、spark シェルで実行すると、結果は次のようになります。

予想通り。エラーは次のとおりです。

また、プログラムでジョブを送信するときに、Spark EC2 クラスターでこの java.io.EOFException を見つけましたが、どのバージョンが必要になるかわかりませんhadoop-client。