問題タブ [apache-spark-2.0]

Java Spark アプリ (実際にはいくつかの計算に RDD を使用します) をDatasetsの代わりに使用するように調整したいと考えていますRDDs。私はデータセットを初めて使用し、対応するデータセット操作にどのトランザクションをマップするかわかりません。

現時点では、次のようにマッピングします。

対応する質問は次のとおりです。

• JavaRDD.mapToPairメソッドに等しいDataset.groupByKey？
• JavaPairRDDにマップしKeyValueGroupedDatasetますか?
• メソッドと等しいメソッドはどれJavaPairRDD.aggregateByKeyですか?

ただし、次の RDD コードをデータセットに移植したいと考えています。

私のコードはまだこれに見えます:

私はspark 2.0.0を使用しています。spark ドライバーからエグゼキューターにパラメーターを渡す方法はありますか? 以下を試しました。

上記のように、パラメーターを渡すカスタム mappartitionfunction を作成しました。partitionfunction の call メソッドで static 変数にアクセスしています。これは、"setmaster("local") を使用してローカルで実行したときに機能しました。しかし、.setmaster("yarn") を使用してクラスターで実行した場合は機能しませんでした。(system.out.println ステートメントで null が出力されます)

ドライバーからエグゼキューターにパラメーターを渡す方法はありますか。

S3 には、snappy 圧縮アルゴリズム (node-snappyパッケージを使用) を使用して圧縮された csv ファイルがいくつかあります。Spark を使用してこれらのファイルを処理しcom.databricks.spark.csvたいのですが、常に無効なファイル入力エラーが発生します。

コード：

エラーメッセージ：

16/09/24 21:57:25 WARN TaskSetManager: ステージ 0.0 でタスク 0.0 が失われました (TID 0、ip-10-0-32-5.ec2.internal): java.lang.InternalError: データを解凍できませんでした。入力が無効です。org.apache.hadoop.io.compress.snappy.SnappyDecompressor.decompressBytesDirect(ネイティブメソッド)で org.apache.hadoop.io.compress.snappy.SnappyDecompressor.decompress(SnappyDecompressor.java:239)で org.apache.hadoop. io.compress.BlockDecompressorStream.decompress(BlockDecompressorStream.java:88) at org.apache.hadoop.io.compress.DecompressorStream.read(DecompressorStream.java:85) at java.io.InputStream.read(InputStream.java:101) org.apache.hadoop.util.LineReader.fillBuffer(LineReader.java:180) で org.apache.hadoop.util.LineReader.readDefaultLine(LineReader.java:216) で org.apache.hadoop.util.LineReader.readLine (LineReader.java:

sparkr.session()私の spark r 1.6 コードは spark2.0 で動作しません。sqlcontext パラメータを渡す代わりに作成するなどの必要な変更を行いましたsparkr.init()…</p>

以下のコードでは、カップルフォルダーからデータフレームにデータをロードしています

動作するspark1.6のread.df

動作しない spark2.0 の read.df

上記の行は次のエラーをスローします。