我正在尝试在Spark 2.0.0中的Dataset[Row]
(DataFrame
(上执行一个简单的映射。 就这么简单
val df: DataSet[Row] = ...
df.map { r: Row => r }
但是编译器抱怨我没有为 map 函数提供隐式Encoder[Row]
参数:
方法映射的参数不足:(隐含证据$7: 编码器[行](。
如果我首先转换为 RDD,一切正常ds.rdd.map { r: Row => r }
但不应该有一种简单的方法来获得Encoder[Row]
,就像元组类型Encoders.product[(Int, Double)]
一样吗?
[请注意,我的Row
是动态调整大小的,因此无法轻松转换为强类型Dataset
。
Encoder
需要知道如何将元素打包在Row
中。因此,您可以使用row.structType
编写自己的Encoder[Row]
,该在运行时确定Row
的元素并使用相应的解码器。
或者,如果您更多地了解进入Row
的数据,则可以使用 https://github.com/adelbertc/frameless/
SSry 有点晚了。希望这对现在遇到问题的人有所帮助。定义编码器的最简单方法是从现有数据帧派生结构:
val df = Seq((1, "a"), (2, "b"), (3, "c").toDF("id", "name")
val myEncoder = RowEndocer(df.schema)
当您需要更改原始数据帧中的现有字段时,此方法可能很有用。
如果你正在处理全新的结构,那么依赖于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
库必须导入。
在映射函数不更改架构的特定情况下,可以传入数据帧本身的编码器:
df.map(r => r)(df.encoder)