問題タブ [spark-avro]

spark 1.3.0 と spark-avro 1.0.0 を使用しています。私のbuild.sbtファイルは次のようになります

アセンブリ プラグインを使用してファット ジャーを作成しています。

ただし、コードが以下のエラーで失敗することがあります。実行するjar -tf Fooassembly.jarと、「com/databricks/spark/avro」フォルダーに多くの .class ファイルが表示されます。ですから、なぜこの特別なクラスに文句を言うのかわかりません。

xml ファイルを avro に変換したい。データは xml 形式になり、最初に kafka トピックにヒットします。次に、flume または spark-streaming を使用して、取り込み、xml から avro に変換し、ファイルを hdfs に格納します。Cloudera 環境があります。

avro ファイルが hdfs にヒットしたときに、後でそれらをハイブ テーブルに読み込む機能が必要です。

これを行うための最良の方法は何ですか？spark-avro などの自動スキーマ変換を試しましたが (これは spark-streaming なしでした)、spark-avro はデータを変換しますが、ハイブはそれを読み取ることができません。Spark avro は、xml をデータフレームに変換してから、データフレームから avro に変換します。avro ファイルは、spark アプリケーションでのみ読み取ることができます。これを正しく使用しているかどうかはわかりません。

avro スキーマの明示的なスキーマを定義する必要があると思います。xml ファイルでこれを行う方法がわかりません。複数の名前空間があり、非常に大規模です。

私は得続けます

java.lang.NoClassDefFoundError: org/apache/avro/mapred/AvroWrapper

show()DataFrame オブジェクトを呼び出すとき。シェル (spark-shell --master yarn) を介してこれを実行しようとしています。DataFrame オブジェクトの作成時にシェルがスキーマを認識していることがわかりますが、データに対してアクションを実行するNoClassDefFoundErrorと、インスタンス化しようとすると常に がスローされますAvroWrapper。avro-mapred-1.8.0.jarクラスターのディレクトリに追加しようとしまし$HDFS_USER/libたが、シェルの起動時に --jar オプションを使用して含めました。これらのオプションはどちらも機能しませんでした。アドバイスをいただければ幸いです。以下はコード例です。

Gobblinを使用して定期的に からリレーショナル データを抽出しOracle、変換してavro公開しています。HDFS

私のdfsディレクトリ構造は次のようになります

私はそれから次のように読み取ろうとしています:

実行するprintSchemaと、スキーマが正しく解釈されていることがわかります。

ただし、countまたはを実行するshowと、DataFramesが空になります。.avroに変換して、ファイルが空でないことを確認しましたJSON

ディレクトリ構造と関係があるのではないかと思います。おそらく、Spark avro ライブラリは、ルートから 1 レベル下の.avroファイルしか参照しません。ログは、t1 の下のディレクトリのみがドライバーにリストされたことを示しているようです。

16/07/07 10:47:09 INFO avro.AvroRelation: リスト hdfs://myhost.mydomain.com:8020/user/username/gobblin/job-output/tables/t1 ドライバー

16/07/07 10:47:09 INFO avro.AvroRelation: リスト hdfs://myhost.mydomain.com:8020/user/username/gobblin/job-output/tables/t1/20160706230001_append ドライバー

誰かが似たようなことを経験したことがありますか、またはこれを回避する方法を知っていますか? t1名前はタイムスタンプによって生成されるため、ディレクトリより下を指さなければなりません。

Avroデータをロードして処理する必要があるsparkプログラムに取り組んでいます。Spark アプリケーションのジョブ ID がどのように作成されるかを理解しようとしています。以下のコード行を使用して、avro データをロードします。

sqlContext.read.format("com.databricks.spark.avro").load(パス)

私の知る限り、プログラムで発生したアクションに基づいてジョブ ID が作成されます。私のジョブは 30 分ごとに実行されるようにスケジュールされています。このアプリケーションの Spark 履歴サーバーを見ると、ロード操作用にジョブ ID が作成されていることがわかります。たまにしか発生せず、ログはまったく問題ないように見えます。私はspark 1.6.1を使用しています

ロード操作がアプリケーションで新しいジョブ ID を作成するかどうか知りたいですか?