斯卡拉星火中的Encoder[Row]

Question

我试图在Spark2.0.0中的Dataset[Row] (DataFrame)上执行一个简单的映射。像这样简单的事情

val df: DataSet[Row] = ...
df.map { r: Row => r }

但是编译器抱怨我没有为map函数提供隐式Encoder[Row]参数：

方法map：(隐式证据$7: EncoderRow)的参数不足。

如果首先转换为RDD，一切都很好，但是不应该有一种像元组类型ds.rdd.map { r: Row => r }那样容易获得Encoder[Row]的方法吗？

请注意，我的Row是动态大小的，因此很难转换为强类型的Dataset。

Answer 1

Encoder需要知道如何在Row中打包元素。因此，您可以使用row.structType编写自己的row.structType，它在运行时确定Row的元素，并使用相应的解码器。

或者，如果您了解更多关于进入Row的数据，则可以使用https://github.com/adelbertc/frameless/。

Answer 2

SSry“有点”迟到了。希望这对正在解决问题的人有帮助。定义编码器的最简单方法是从现有的DataFrame派生结构：

val df = Seq((1, "a"), (2, "b"), (3, "c").toDF("id", "name")
val myEncoder = RowEndocer(df.schema)

当您需要更改原始DataFrame中的现有字段时，这种方法可能很有用。

如果您处理的是全新的结构，那么使用StructType和StructField进行明确的定义(如@Reactormonk的小神秘响应中所建议的那样)。

定义相同编码器的示例：

val myEncoder2 = RowEncoder(StructType(
  Seq(StructField("id", IntegerType), 
      StructField("name", StringType)
  )))

请记住，必须导入org.apache.spark.sql._、org.apache.spark.sql.types._和org.apache.spark.sql.catalyst.encoders.RowEncoder库。

Answer 3

在映射函数不更改模式的特定情况下，可以传递DataFrame本身的编码器：

df.map(r => r)(df.encoder)