问题标签 [apache-spark-dataset]

我仍在努力理解最近推出的 Spark 数据集的全部功能。

是否有关于何时使用 RDD 和何时使用数据集的最佳实践？

Databricks在他们的公告中解释说，通过使用数据集，可以实现运行时和内存的惊人减少。仍然有人声称数据集被设计为“与现有的 RDD API 一起工作”。

这只是对向下兼容性的引用，还是在某些情况下人们更愿意使用 RDD 而不是数据集？

我正在尝试使用 Spark Dataset API，但在进行简单连接时遇到了一些问题。

假设我有两个带有 fields: 的数据集date | value，那么在DataFrame我的 join 情况下将如下所示：

但是对于Dataset有.joinWith方法，但相同的方法不起作用：

所需的论据是.joinWith什么？

根据介绍 Spark 数据集：

由于我们对 Spark 2.0 的期待，我们计划对 Datasets 进行一些激动人心的改进，特别是： ... 自定义编码器——虽然我们目前为多种类型自动生成编码器，但我们希望为自定义对象开放一个 API。

并尝试存储自定义类型Dataset导致以下错误，例如：

找不到存储在数据集中的类型的编码器。通过导入 sqlContext.implicits 支持原始类型（Int、String 等）和产品类型（案例类）。未来版本中将添加对序列化其他类型的支持

或者：

Java.lang.UnsupportedOperationException：没有找到用于 .... 的编码器

是否有任何现有的解决方法？

请注意，此问题仅作为社区 Wiki 答案的入口点存在。随时更新/改进问题和答案。

有谁可以通过示例帮助我了解 DataSet API 和 DataFrame API 之间的区别？为什么需要在 Spark 中引入 DataSet API？

我写过two case class扩展 Base abstract class。我有每个班级的两个列表（listA和listB）。当我想合并这两个列表时，我无法将最终列表转换为 Apache Spark 1.6.1 数据集。

Apache Spark 将引发此异常：

当我想从listSpark 创建 RDD 时不会抛出任何异常，但是当我使用toDS()方法将 RDD 转换为 Dataset 时，会抛出这个先前的异常。

我RDD[LabeledPoint]打算在机器学习管道中使用。我们如何将其转换RDD为DataSet？请注意，较新的 spark.mlapi 需要Dataset格式中的输入。

Spark SQL DataFrame/Dataset 执行引擎有几个非常有效的时间和空间优化（例如 InternalRow 和表达式 codeGen）。根据许多文档，对于大多数分布式算法来说，它似乎是比 RDD 更好的选择。

但是，我做了一些源代码研究，仍然不相信。我毫不怀疑 InternalRow 更紧凑，可以节省大量内存。但是算法的执行可能不会比保存预定义的表达式更快。也就是说，在 的源代码中指出 org.apache.spark.sql.catalyst.expressions.ScalaUDF，每个用户定义的函数都会做 3 件事：

1. 将催化剂类型（用于 InternalRow）转换为 scala 类型（用于 GenericRow）。
2. 应用函数
3. 将结果从 scala 类型转换回催化剂类型

显然，这甚至比直接在 RDD 上应用该函数而不进行任何转换还要慢。任何人都可以通过一些真实案例分析和代码分析来确认或否认我的猜测吗？

非常感谢您的任何建议或见解。

Spark 数据集从 Row's 转移到EncoderPojo's/primitives 的's。Catalyst引擎使用ExpressionEncoder转换 SQL 表达式中的列。但是，似乎没有其他子类Encoder可用作我们自己的实现的模板。

下面是一个在 Spark 1.X / DataFrames 中没有在新机制下编译的代码示例：

}

我们得到一个编译器错误

那么，不知何故/某处应该有办法

• 定义/实现我们的自定义编码器
• DataFrame在（现在是类型的 Dataset Row）上执行映射时应用它
• 注册编码器以供其他自定义代码使用

我正在寻找成功执行这些步骤的代码。

我正在使用 Spark 数据集（Spark 1.6.1 版本）。下面是我的代码

现在我想在多个列上执行 group by 子句？怎么做？ result.groupBy(_._1._1.created_at)这样我可以吗？如果是，那么我无法将结果视为一个组，如何在多列上执行此操作？

我正在使用带有 Spark 2.0 预览版的 Databricks 社区版。我尝试了以下（简单）代码：

我收到以下错误消息：

错误：类型不匹配；
找到：org.apache.spark.sql.TypedColumn[C1,Seq[C1]]
需要：org.apache.spark.sql.TypedColumn[C1,Seq[C1]]
val g_c1 = teams.groupByKey(_.f1)。聚合序列[C1]

当我使用

我得到：

错误：类型不匹配；
找到：org.apache.spark.sql.TypedColumn[C1,Seq[C1]]
需要：org.apache.spark.sql.TypedColumn[C1,?]
val g_c1 = teams.groupByKey(_.f1).aggSeq[C1 ]

有什么提示吗？