別のデータセットから間隔を取得して新しいデータセットを作成しようとしています。たとえば、dataset1 を入力、dataset2 を出力とします。

配列を使用してそれを行うことができましたが、mlib にはデータセットが必要です。

配列を使用した私のコード:

FlatMap は進むべき道のように思えますが、関数はどのようにしてデータセット内の次の値を検索できるのでしょうか?

spark 2.0 データセット API で日時の解析を適切に実行するにはどうすればよいですか?

データフレーム/RDDライクなサンプルがたくさんあります

• スパーク日付の解析
• Spark で文字列フィールドをタイムスタンプに変換するより良い方法
• データフレームで列の型を文字列から日付に変更するには?

みたいなクラス

型をキャストするには十分ではありません。データセット API を使用してこれを適切に実行するにはどうすればよいですか?

編集

データのロードは機能します。例えば。aprint schemaは示していますmyDateFiled: timestamp (nullable = true) が、myDf.show の結果は

これにより、日付の解析が間違っていると信じるようになりました。どうすればいいの？

Hadoop 2.7.2、Centos 7.2 で Apache Spark 2.0 を実行しているクラスターがあります。Spark DataFrame/DataSet API を使用していくつかの新しいコードを作成しましたが、Windows Azure Storage Blob (デフォルトの HDFS の場所) にデータを書き込んでから読み取った後、結合で誤った結果が生じることに気付きました。クラスターで実行されている次のコード スニペットで問題を再現できました。

出力

どちらが正しい。ただし、データの書き込みと読み取りの後、これが表示されます

出力

ただし、RDD API を使用すると正しい結果が得られます

出力形式を寄木細工ではなく ORC に変更しようとしましたが、同じ結果が得られます。クラスタではなくローカルで Spark 2.0 を実行すると、この問題は発生しません。また、Hadoop クラスターのマスター ノードでローカル モードで spark を実行することもできます。YARN の上で実行している場合にのみ、この問題が発生します。

これもこの問題と非常によく似ているようです: https://issues.apache.org/jira/browse/SPARK-10896

この記事DataFrameでは、Spark の は と同等であると主張していますDataset[Row]が、このブログ投稿DataFrameでは にスキーマがあることを示しています。

RDD をaに変換するブログ投稿の例を見てみましょうDataFrame。DataFrameDataset[Row]RDDDataFrame

しかし、代わりにそれがこれであることを示しています

明らかに、データフレームは実際には行とスキーマのデータセットです。

ウィンドウ関数を使用してウィンドウ内の平均を計算しようとしていますが、計算前に 0 を除外しています。

このようにサイズをフィルタリングしようとしましたが、うまくいきません。

ついに、

これを解決する方法はありますか？

コンパイル時の型の安全性により、Spark 2.0 DataSet に非常に満足しています。しかし、ここで解決できない問題がいくつかあります。これに関する適切なドキュメントも見つかりませんでした。

問題 #1 - 集計された列での除算操作 - 以下のコードを検討してください - DataSet[MyCaseClass] があり、c1、c2、c3 および sum(c4) / 8 で groupByKey を使用したいと考えていました。 sum ですが、divide(8) のコンパイル時エラーが発生します。どうすれば次のことを達成できるのだろうか。

.divide(8) 操作を削除して上記のコマンドを実行すると、以下の出力が得られます。

問題 2 - groupedByKey の結果を別の Typed DataFrame に変換する - 問題 の 2 番目の部分は、再度 Typed DataSet を出力したいということです。そのために、別のケースクラスがあります（必要かどうかはわかりません）が、グループ化された結果でマップする方法がわかりません-

しかし、キーの結果によってグループ化された結果がAnotherClassに直接マップされていないため、これも例外で失敗します。

PS：上記を達成するための他の解決策は大歓迎です。

別の行に基づいていくつかの行を更新する方法を理解しようとしています。

たとえば、次のようなデータがあります

同じ都市のユーザーを同じ groupId (1 または 2) に更新したい

RDD または Dataset でこれを達成するにはどうすればよいですか?

完全を期すために、Idが String の場合、dense ランクは機能しないのでしょうか?

例えば ​​？

したがって、結果は次のようになります。

Row RDD を Typed RDD に変換することは可能ですか? 以下のコードで、行 JavaRDD をカウンター型 JavaRDD に変換できますか?

コード:

スカラでは：

ケースクラスにある種のコンストラクターを提供する必要がありますか?

サード パーティのライブラリで定義されている既存のドメイン オブジェクト、つまり HAPI-FHIR のPatientオブジェクトを使用して、次のように厳密に型指定された Spark を作成しようとしていますDataSet[Patient]。

しかし、上記の RDD#toDS 呼び出しを行うと、非常に長いStackOverflowError.

完全なスタック トレースはこちら: https://gist.github.com/vratnagiri-veriskhealth/6dcec9dbc6f74308019ab16c8d278a9b

上で参照したドメイン オブジェクトの複雑さを考えると、これはばかげた用事かもしれないと思いますが、私が scala の初心者であることを考えると、これを取得する可能性のある簡単な調整を見逃していないことを確認したいと思います。私がこの追求を放棄する前に働いています。

スタックトレースの一部は次のとおりです。

ありがとう！