問題タブ [apache-spark-dataset]

最近導入された Spark Datasets の全機能を理解するのにまだ苦労しています。

RDD を使用する場合とデータセットを使用する場合のベスト プラクティスはありますか?

Databricksは発表の中で、Datasets を使用することで、ランタイムとメモリの両方を大幅に削減できると説明しています。それでも、データセットは「既存の RDD API と連携して動作する」ように設計されていると主張されています。

これは単なる下位互換性への言及ですか、それともデータセットよりも RDD を使用することを好むシナリオはありますか?

例を使って DataSet API と DataFrame API の違いを理解するのを手伝ってくれる人はいますか? Spark に DataSet API を導入する必要があったのはなぜですか?

two case classBase を拡張することを書きましたabstract class。各クラスの 2 つのリスト (listAおよびlistB) があります。これら 2 つのリストをマージしたい場合、最終的なリストを Apache Spark 1.6.1 データセットに変換できません。

Apache Spark では、次の例外が発生します。

Sparkから RDD を作成したい場合list、例外はスローされませんが、toDS()メソッドを使用して RDD をデータセットに変換すると、この前の例外がスローされます。

RDD[LabeledPoint]機械学習パイプライン内で使用することを意図しています。RDDそれを に変換するにはどうすればよいDataSetでしょうか。新しい API では、この形式spark.mlの入力が必要であることに注意してください。Dataset

Spark SQL DataFrame/Dataset 実行エンジンには、非常に効率的な時間と空間の最適化がいくつかあります (InternalRow と式 codeGen など)。多くのドキュメントによると、ほとんどの分散アルゴリズムでは RDD よりも優れたオプションのようです。

ただし、ソースコードを調査しましたが、まだ確信が持てません。InternalRow がはるかにコンパクトであり、大量のメモリを節約できることは間違いありません。しかし、アルゴリズムの実行は、定義済みの式を保存するよりも高速ではない場合があります。org.apache.spark.sql.catalyst.expressions.ScalaUDFつまり、 のソースコードでは、すべてのユーザー定義関数が次の 3 つのことを行うことが示され ています。

1. Catalyst タイプ (InternalRow で使用) を scala タイプ (GenericRow で使用) に変換します。
2. 関数を適用する
3. 結果をscala型からcatalyst型に戻す

どうやらこれは、関数を変換せずに RDD に直接適用するよりもさらに遅いようです。実際のケースのプロファイリングとコード分析によって、誰かが私の推測を確認または否定できますか?

提案や洞察をありがとうございました。

Spark データセットは、Row からEncoderPojo/プリミティブの に移動します。Catalystエンジンは を使用して、SQL 式のExpressionEncoder列を変換します。Encoderただし、独自の実装のテンプレートとして使用できる他のサブクラスはないようです。

これは、Spark 1.X / DataFrames で問題なく、新しい体制でコンパイルされないコードの例です。

}

のコンパイラエラーが発生します

したがって、どういうわけか/どこかに手段があるはずです

• カスタム Encoder を定義/実装する
• でマッピングを実行するときにそれを適用しますDataFrame(現在はタイプのデータセットですRow)
• 他のカスタム コードで使用するエンコーダーを登録します。

これらの手順を正常に実行するコードを探しています。

Spark 2.0 プレビューで Databricks Community Edition を使用しています。次の（単純な）コードを試しました：

次のエラー メッセージが表示されます。

エラー: タイプが一致しません。
見つかった: org.apache.spark.sql.TypedColumn[C1,Seq[C1]]
必須: org.apache.spark.sql.TypedColumn[C1,Seq[C1]]
val g_c1 = teams.groupByKey(_.f1)。 aggSeq[C1]

私が使うとき

私は得る：

エラー: タイプが一致しません。
見つかった: org.apache.spark.sql.TypedColumn[C1,Seq[C1]]
必須: org.apache.spark.sql.TypedColumn[C1,?]
val g_c1 = teams.groupByKey(_.f1).aggSeq[C1 ]

ヒントはありますか？