問題タブ [spark-csv]

上記のコードを使用してデータフレームを作成すると、次のエラーが発生します。

ERROR Executor: ステージ 1.0 (TID 1) のタスク 0.0 での例外/言語/オブジェクト;

parseLib オプションを回避すると、問題なく動作します。spark-csv パーサーを Univocity に設定したいと同時に、inferSchema を使用して spark csv によってデータ型が認識される必要があります。

注: 私は spark-csv 1.3 を使用しています (どのバージョンでも動作しません) Spark: 1.6.2 Scala: 2.10.5

ありがとう。

Scala 2.11 で spark-core バージョン 2.0.1 を使用しています。\エスケープを持つcsvファイルを読み取る簡単なコードがあります。

ドキュメントによると、\ は csv リーダーのデフォルトのエスケープです。しかし、うまくいきません。Spark はデータの一部として \ を読み取ります。例: csv ファイルの City 列はnorth rocks\,auです。私は city 列がnorthrocks,auとしてコードで読み取られることを期待しています。しかし代わりに、spark はそれをnorthrocks\として読み取り、 auを次の列に移動します。

私は次のことを試しましたが、うまくいきませんでした:

• 明示的に定義されたエスケープ .option("escape","\")
• エスケープを | に変更 または : ファイル内およびコード内
• spark-csv ライブラリを使用してみました

同じ問題に直面している人はいますか? 何か不足していますか？

ありがとう

でEC2セットアップしていr3.8xlarge (32 cores, 244G RAM)ます。

私のSparkアプリケーションでは、DataBrickからS3使用して 2 つの csv ファイルを読み取っています。各 csv には約 500 万行あります。Spark-CSV私はunionAll2 つの DataFrame であり、結合された DataFrame で実行していdropDuplicatesます。

しかし、私が持っているとき、

スパークはより遅い.setMaster("local")

32コアだともっと速くなるんじゃない？

json ファイルからの情報を使用してカスタム org.apache.spark.sql.types.StructType スキーマ オブジェクトを作成する必要があります。json ファイルは何でもかまいません。そのため、プロパティ ファイル内でパラメトリック化しました。

プロパティファイルは次のようになります。

generated_schema.json ファイルは次のようになります。

だから、これは私がそれを解決できると思った方法です：

コードが最後の行 .parquet(pathParquet) を実行すると、例外が発生します。

このコードの出力は次のようになります。

schema_json オブジェクトと myDF.schema.json オブジェクトの内容は同じでなければなりません。しかし、それは起こりませんでした。これでエラーが発生するはずだと思います。

最後に、この例外でジョブがクラッシュします。

実際には、json スキーマ ファイルを提供しない場合、ジョブは正常に実行されますが、このスキーマでは...

誰でも私を助けることができますか？csv ファイルと json スキーマ ファイルから始めて、いくつかの寄木細工のファイルを作成したいだけです。

ありがとうございました。

依存関係は次のとおりです。

アップデート

未解決の問題があることがわかりますが、

https://github.com/databricks/spark-csv/issues/61

240GB の「大きな」EC2 ボックスで 3 つのノードを使用してスタンドアロン モードで Spark を実行し、DataFrames に読み込まれた 3 つの CSV ファイルを JavaRDD にマージして、s3a を使用して S3 の CSV パーツ ファイルを出力します。

Spark UI から、最初の段階の読み取りとマージによって、期待どおり 100% の CPU で実行される最終的な JavaRDD を生成することがわかりますが、最終段階の CSV ファイルとしての書き込みsaveAsTextFile at package.scala:179は、3 つのノードのうちの 2 つで何時間も「スタック」します。 32 のタスクのうち 2 つは数時間かかります (ボックスは全体で CPU 6%、メモリ 86%、ネットワーク IO 15kb/s、ディスク IO 0 です)。

3 つの入力 DataFrame のそれぞれで再パーティション 16 を使用して、圧縮されていない CSV を読み書きしています (圧縮されていない CSV は gzip 圧縮された CSV よりもはるかに高速であることがわかりました)。

スタンドアロン ローカル クラスタの 3 つのノードのうち 2 つのノードでほとんど何もしないのに、なぜ最終段階で何時間もかかるのかについて調査できるヒントをいただければ幸いです。

どうもありがとう

- - アップデート - -

s3a ではなくローカル ディスクに書き込もうとしましたが、症状は同じです。最終段階の 32 のタスクのうち 2 つがsaveAsTextFile何時間も「スタック」します。

カスタム Estimator の transformSchema メソッドの場合、入力データ フレームのスキーマをケース クラスで定義されたスキーマと比較できる必要があります。通常、これは、以下に概説するように、ケース クラスから Spark StructType / Schema を生成するように実行できます。ただし、間違った nullability が使用されています。

によって推論される df の実際のスキーマは次のspark.read.csv().as[MyClass]ようになります。

そしてケースクラス：

比較するには、次を使用します。

残念ながらfalse、ケース クラスから手動で推論された新しいスキーマが nullable に設定されているため、これは常に生成されますtrue(ja java.Integer が実際には null である可能性があるため)。

nullable = falseスキーマの作成時にどのように指定できますか?

Pythonでspark sqlデータフレームをエクスポートする方法を理解するために、次のリンクを参照しました

• https://github.com/databricks/spark-csv
• Spark SQL から CSV にデータをエクスポートする方法

私のコード:

マスターURLに次のjarを渡すspark-submitでジョブをロードします

次のエラーが表示されます