問題タブ [apache-spark-dataset]

そのため、データセット API を使用するために取り組んでいるプロジェクトを再フォーマットしようとしており、エンコード エラーに関する問題が発生しています。私が読んだことから、プリミティブ値の配列をデータセットに格納できるはずだと思います。ただし、次のクラスではエンコード エラーが発生します。

誰かがここで何が問題なのかを理解するのを手伝ってくれませんか? データセットは現在、プリミティブの配列を処理できませんか、それともそれらを機能させるために何か追加する必要がありますか?

ありがとうございました

編集1：

ここに私が得ている完全なエラーがあります

spark 2.0.1-SNAPSHOT で新しいストリーミング書き込み機能を使用しようとしています。結果を保持するために実際にサポートされている出力データソースはどれですか? 次のような出力をコンソールに表示できました。

しかし、.format("console") を "json" または "jdbc" に変更しようとすると、次のメッセージが表示されます: データ ソース xxx はストリーミング書き込みをサポートしていません。

だからこれは私が試してきたことです：

SparseMatrix を列の 1 つとして返すことを使用して集計を実行しmapGroups、列を合計しようとしています。

case class列名を提供するために、マップされた行のスキーマを作成しました。行列の列は と入力されorg.apache.spark.mllib.linalg.Matrixます。toDF集計を実行する前に実行しないと( select(sum("mycolumn"))、タイプの不一致エラーが 1 つ発生します ( required: org.apache.spark.sql.TypedColumn[MySchema,?])。含めるtoDFと、別のタイプの不一致エラーが発生します: cannot resolve 'sum(mycolumn)' due to data type mismatch: function sum requires numeric types, not org.apache.spark.mllib.linalg.MatrixUDT. それで、それを行う正しい方法は何ですか？

データセット スキーマを作成する方法を理解するのに苦労しています。1 つの列にキーのタプルがあり、2 番目の列に集計がある集計からのデータセットがあります。

このスキーマを適用できないのはなぜですか?

Scala を反映しようとしても失敗しましたcase class:

spark DataSetの関数を見ると、select()さまざまな関数シグネチャが生成されています。

これは、 MyClass のメンバーを直接参照してタイプ セーフにする必要があることを示唆しているようですが、方法がわかりません...

ds.select("member")ds.select(_.member)もちろん動作します..何とか動作するようです?

私は Apache Spark 2.0 を使用しておりcase class、言及用のスキーマを作成していますDetaSet。How to store custom objects in Dataset?に従ってカスタム エンコーダーを定義しようとすると、、java.time.LocalDate次の例外が発生したため：

以下はコードによるものです：

Spark 用のサードパーティ API のエンコーダーを定義するにはどうすればよいですか?

アップデート

ケース クラス全体のエンコーダを作成するときはdf.map..、次のようにオブジェクトをバイナリにマップします。

FireService のマップを期待していますが、マップのバイナリを返します。