如何在Spark 2.X数据集中创建自定义编码器？

Question

Spark数据集从Row转向使用Encoder来处理Pojo和原始类型。Catalyst引擎使用ExpressionEncoder将SQL表达式中的列进行转换。然而，似乎没有其他可用的Encoder子类可用于作为我们自己实现的模板。

这是一个在Spark 1.X / DataFrames中可以正常工作但在新的体系结构中无法编译的代码示例：

//mapping each row to RDD tuple
df.map(row => {
    var id: String = if (!has_id) "" else row.getAs[String]("id")
    var label: String = row.getAs[String]("label")
    val channels  : Int = if (!has_channels) 0 else row.getAs[Int]("channels")
    val height  : Int = if (!has_height) 0 else row.getAs[Int]("height")
    val width : Int = if (!has_width) 0 else row.getAs[Int]("width")
    val data : Array[Byte] = row.getAs[Any]("data") match {
      case str: String => str.getBytes
      case arr: Array[Byte@unchecked] => arr
      case _ => {
        log.error("Unsupport value type")
        null
      }
    }
    (id, label, channels, height, width, data)
  }).persist(StorageLevel.DISK_ONLY)

我们得到了一个编译器错误

Error:(56, 11) Unable to find encoder for type stored in a Dataset.
Primitive types (Int, String, etc) and Product types (case classes) are supported 
by importing spark.implicits._  Support for serializing other types will be added in future releases.
    df.map(row => {
          ^

所以，必须有某种方式/某个地方可以：

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

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

- WestCoastProjects

---

可能是如何在数据集中存储自定义对象的重复问题。 - Alec

看起来他们在3.2中添加了支持：https://issues.apache.org/jira/browse/SPARK-23862 - Geoff Langenderfer

Answer 1

据我所知，自1.6版本以来并没有什么实质性的变化，因此在How to store custom objects in Dataset?中描述的解决方案是唯一可用的选项。尽管如此，对于产品类型，默认编码器应该可以很好地工作。

要了解为什么你的代码在1.x版本中可以工作但在2.0.0版本中可能无法工作，你需要检查签名。在1.x版本中，DataFrame.map是一个方法，它接受函数Row => T并将RDD[Row]转换为RDD[T]。

在2.0.0版本中，DataFrame.map也接受类型为Row => T的函数，但它将Dataset[Row]（又名DataFrame）转换为Dataset[T]，因此T需要一个Encoder。如果您想获得“旧”的行为，应明确使用RDD：

df.rdd.map(row => ???)

有关 Dataset[Row] 的 map 操作请参考 Encoder error while trying to map dataframe row to updated row

- zero323

Answer 2

你导入了隐式编码器吗？

import spark.implicits._

http://spark.apache.org/docs/2.0.0-preview/api/scala/index.html#org.apache.spark.sql.Encoder

- eyal edelman

Answer 3

我导入了spark.implicits._，其中spark是SparkSession，这解决了错误并成功导入了自定义编码器。

此外，编写自定义编码器也是一种解决方法，但我还没有尝试过。

工作解决方案：创建SparkSession并导入以下内容

import spark.implicits._

- Valan Aravind