問題タブ [apache-spark-1.6]

pySpark を使用して寄木細工のファイルを作成します。そのファイルの hdfs ブロック サイズを変更したいと思います。私はこのようにブロックサイズを設定しましたが、うまくいきません:

これは、pySpark ジョブを開始する前に設定する必要がありますか? もしそうなら、それを行う方法。

Kafka をデータ ソースとして Spark Streaming アプリケーションを実行したいと考えています。ローカルでは問題なく動作しますが、クラスターでは失敗します。私はspark 1.6.2とScala 2.10.6を使用しています。

ソース コードとスタック トレースは次のとおりです。

DevMain.scala

object DevMain は Logging を使用して App を拡張します {

6 行目に a がNullPointerException表示され、コードが入力されませんlme.parser。

これはlme.parser次のとおりです。

の line1 にログオンしましたがlme.parser、印刷されず、入力されませんlem.parser。

... 3 つ以上

これは、ブロードキャスト変数が変更された後の新しい例外です

私が持っているデータを処理するために、以前にスキーマを抽出しているので、データセットを読み取るときに、スキーマを推測するという高価な手順を実行する代わりにスキーマを提供します。

スキーマを構築するには、いくつかの異なるスキーマを最終的なスキーマにマージする必要があるため、メソッドunion (++)とdistinctメソッドを使用していますが、org.apache.spark.sql.AnalysisException: Duplicate column(s)例外が発生し続けます。

たとえば、次の構造に 2 つのスキーマがあるとします。

出力：

別のスキーマと完全に一致するスキーマ構造のみが によって除外されることを理解していますdistinct。ただし、結果を次のようにしたい:

すべてが 1 つのスキーマに「結合」されます。scala ドキュメントのすべてのメソッドをふるいにかけましたが、これを解決する正しいメソッドが見つからないようです。何か案は？

編集：

最終的な目標は、メソッドを使用して JSON 文字列の RDD にフィードしfinal_schema、sqlContext.read.schema読み取るreadことです。